diff --git a/.gitea/workflows/ci.yaml b/.gitea/workflows/ci.yaml index 2480c13..260eced 100644 --- a/.gitea/workflows/ci.yaml +++ b/.gitea/workflows/ci.yaml @@ -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] diff --git a/edu_master/.env.example b/edu_master/.env.example index ec9c514..5017abe 100644 --- a/edu_master/.env.example +++ b/edu_master/.env.example @@ -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 diff --git a/edu_master/phpsessid-bot/bot.py b/edu_master/phpsessid-bot/bot.py index 963a143..3e1915e 100644 --- a/edu_master/phpsessid-bot/bot.py +++ b/edu_master/phpsessid-bot/bot.py @@ -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}') diff --git a/tests/test_session_keeper.py b/tests/test_session_keeper.py new file mode 100644 index 0000000..8f97c51 --- /dev/null +++ b/tests/test_session_keeper.py @@ -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()