fix(edu-master): bound session refreshes and expire stale cookies

2026-10-06 15:58:44 +02:00
5 changed files with 111 additions and 47 deletions
@@ -141,6 +141,7 @@ jobs:
          export PATH="$tools_dir:$PATH"
          ruff check .
          ruff format --check .
+          python3 -m unittest discover -s tests -v

  lint-yaml:
    runs-on: [self-hosted, linux, arch, homelab]
@@ -1,10 +1,11 @@
-EDU_LOGIN=your_edu_login_here
-EDU_PASSWORD=your_edu_password_here
-EDU_URL_LOGIN=https://edu.edu.vn.ua/user/login
-EDU_URL_VERIFY=https://edu.edu.vn.ua/course/userlist
-PHPSESSID_INTERVAL=10
+KEEPER_LOGIN=your_edu_login_here
+KEEPER_PASSWORD=your_edu_password_here
+EDU_URL_BASE=https://edu.edu.vn.ua
+EDU_URL_LOGIN=/user/login
+EDU_URL_COURSES=/course/userlist
+KEEPER_INTERVAL=10
 USER_AGENT="Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/142.0.0.0 Safari/537.36"
-WEBINAR_URL=https://edu.edu.vn.ua/webinar/useractive
+EDU_URL_WEBINAR=/webinar/useractive
 WEBINAR_CHECK_INTERVAL=60
 REDIS_HOST=redis
 REDIS_PORT=6379
@@ -48,17 +48,43 @@ def touch_success_file():
        logger.error(f'Failed to touch success file: {e}')


+def refresh_session(session, redis_client):
+    """Publish a verified cookie with a lifetime tied to the refresh interval."""
+    login_response = session.post(
+        URL_LOGIN, data={'login': LOGIN, 'password': PASSWORD}, allow_redirects=True, timeout=(10, 30)
+    )
+    login_response.raise_for_status()
+    verify_response = session.get(URL_VERIFY, allow_redirects=False, timeout=(10, 30))
+    if verify_response.status_code != 200:
+        logger.warning('Session verification failed (HTTP %s)', verify_response.status_code)
+        return False
+    phpsessid = session.cookies.get('PHPSESSID')
+    if not phpsessid:
+        logger.warning('Verified response did not provide a PHPSESSID cookie')
+        return False
+    redis_client.set('EDU_PHPSESSID', phpsessid, ex=INTERVAL * 120)
+    touch_success_file()
+    logger.info('Verified session saved to Redis')
+    return True
+
+
 def main():
+    if not LOGIN or not PASSWORD:
+        raise ValueError('KEEPER_LOGIN and KEEPER_PASSWORD must be set')
+    if INTERVAL <= 0:
+        raise ValueError('KEEPER_INTERVAL must be a positive number of minutes')
    logger.info('Starting Session Keeper Bot')

    # Connect to Redis
    try:
-        redis_client = redis.Redis(host=REDIS_HOST, port=REDIS_PORT, decode_responses=True)
+        redis_client = redis.Redis(
+            host=REDIS_HOST, port=REDIS_PORT, decode_responses=True, socket_connect_timeout=5, socket_timeout=5
+        )
        redis_client.ping()
        logger.info(f'Connected to Redis at {REDIS_HOST}:{REDIS_PORT}')
    except Exception as e:
        logger.error(f'Failed to connect to Redis: {e}')
-        return
+        raise

    session = requests.Session()

@@ -83,44 +109,7 @@ def main():

    while True:
        try:
-            logger.info('Attempting login...')
-
-            # Login payload
-            payload = {'login': LOGIN, 'password': PASSWORD}
-
-            # Perform Login
-            # Note: The user request shows a POST to /user/login with form data
-            # We need to make sure we handle the PHPSESSID correctly.
-            # If we already have a PHPSESSID, requests will send it.
-
-            login_response = session.post(URL_LOGIN, data=payload, allow_redirects=True)
-
-            logger.info(f'Login Response Status: {login_response.status_code}')
-            logger.info(f'Cookies after login: {session.cookies.get_dict()}')
-
-            # Verify Session
-            logger.info('Verifying session...')
-            verify_response = session.get(URL_VERIFY, allow_redirects=False)
-
-            logger.info(f'Verify Response Status: {verify_response.status_code}')
-
-            if verify_response.status_code == 200:
-                logger.info('Session verification SUCCESS (200 OK).')
-                touch_success_file()
-
-                # Save PHPSESSID to Redis
-                phpsessid = session.cookies.get('PHPSESSID')
-                if phpsessid:
-                    try:
-                        redis_client.set('EDU_PHPSESSID', phpsessid)
-                        logger.info(f'Saved PHPSESSID to Redis: {phpsessid}')
-                    except Exception as e:
-                        logger.error(f'Failed to save PHPSESSID to Redis: {e}')
-            elif verify_response.status_code == 302:
-                logger.warning('Session verification FAILED (302 Redirect). Session might be invalid.')
-            else:
-                logger.warning(f'Session verification returned unexpected status: {verify_response.status_code}')
-
+            refresh_session(session, redis_client)
        except Exception as e:
            logger.error(f'An error occurred: {e}')

@@ -1,7 +1,6 @@
 POSTGRES_ADMIN_PASSWORD=
 AUTHENTIK_DB_PASSWORD=
 GITEA_DB_PASSWORD=
-NETBOX_DB_PASSWORD=
 NETRONOME_DB_PASSWORD=
 PENPOT_DB_PASSWORD=
 STATUSPAGE_DB_PASSWORD=
@@ -0,0 +1,74 @@
+"""Session publication checks without Redis, the EDU website, or credentials."""
+
+import importlib.util
+import sys
+import unittest
+from pathlib import Path
+from unittest.mock import Mock, patch
+
+SCRIPT = Path(__file__).resolve().parents[1] / 'edu_master/phpsessid-bot/bot.py'
+spec = importlib.util.spec_from_file_location('session_keeper', SCRIPT)
+bot = importlib.util.module_from_spec(spec)
+with patch.dict(sys.modules, {'redis': Mock(), 'requests': Mock()}):
+    spec.loader.exec_module(bot)
+
+
+class SessionKeeperTests(unittest.TestCase):
+    def setUp(self):
+        self.session = Mock()
+        self.session.get.return_value.status_code = 200
+        self.session.cookies.get.return_value = 'test-cookie'
+        self.redis = Mock()
+        self.touch = patch.object(bot, 'touch_success_file').start()
+        self.addCleanup(patch.stopall)
+
+    def test_verified_cookie_expires_and_is_not_logged(self):
+        with self.assertLogs(bot.logger, level='INFO') as logs:
+            self.assertTrue(bot.refresh_session(self.session, self.redis))
+        self.redis.set.assert_called_once_with('EDU_PHPSESSID', 'test-cookie', ex=bot.INTERVAL * 120)
+        self.touch.assert_called_once()
+        self.assertNotIn('test-cookie', '\n'.join(logs.output))
+        self.assertEqual(self.session.post.call_args.kwargs['timeout'], (10, 30))
+        self.assertEqual(self.session.get.call_args.kwargs['timeout'], (10, 30))
+
+    def test_redirect_does_not_publish(self):
+        self.session.get.return_value.status_code = 302
+        self.assertFalse(bot.refresh_session(self.session, self.redis))
+        self.redis.set.assert_not_called()
+        self.touch.assert_not_called()
+
+    def test_missing_cookie_does_not_mark_success(self):
+        self.session.cookies.get.return_value = None
+        self.assertFalse(bot.refresh_session(self.session, self.redis))
+        self.redis.set.assert_not_called()
+        self.touch.assert_not_called()
+
+    def test_redis_failure_does_not_mark_success(self):
+        self.redis.set.side_effect = OSError('redis unavailable')
+        with self.assertRaises(OSError):
+            bot.refresh_session(self.session, self.redis)
+        self.touch.assert_not_called()
+
+    def test_http_timeout_does_not_publish(self):
+        self.session.post.side_effect = TimeoutError('EDU unavailable')
+        with self.assertRaises(TimeoutError):
+            bot.refresh_session(self.session, self.redis)
+        self.redis.set.assert_not_called()
+        self.touch.assert_not_called()
+
+    def test_missing_credentials_fail_before_network_access(self):
+        with patch.object(bot, 'LOGIN', None), self.assertRaises(ValueError):
+            bot.main()
+
+    def test_nonpositive_interval_fails_before_network_access(self):
+        with (
+            patch.object(bot, 'LOGIN', 'test'),
+            patch.object(bot, 'PASSWORD', 'test'),
+            patch.object(bot, 'INTERVAL', 0),
+            self.assertRaises(ValueError),
+        ):
+            bot.main()
+
+
+if __name__ == '__main__':
+    unittest.main()