From c4cef0d9b51c3f1d15cac29af12265b2d4eec95e Mon Sep 17 00:00:00 2001 From: b267a Date: Thu, 22 Jan 2026 16:17:05 +0100 Subject: [PATCH] Refactor SSO authentication to support multiple providers and enhance error handling --- .gitignore | 4 ++- README.md | 72 ++++++++++++++++++++++++++++++-------- auth/providers/__init__.py | 16 +++++++++ auth/providers/base.py | 35 ++++++++++++++++++ auth/providers/kit.py | 36 +++++++++++++++++++ auth/session.py | 52 +++++++++++++-------------- booking/client.py | 34 +++++++++++++++--- config/constants.py | 4 ++- main.py | 43 ++++++++++++++--------- utils/helpers.py | 3 +- 10 files changed, 233 insertions(+), 66 deletions(-) create mode 100644 auth/providers/__init__.py create mode 100644 auth/providers/base.py create mode 100644 auth/providers/kit.py diff --git a/.gitignore b/.gitignore index cff7ac8..eeb365b 100644 --- a/.gitignore +++ b/.gitignore @@ -1,2 +1,4 @@ /.idea -*.env \ No newline at end of file +*.env +*pyc +__pycache__/ \ No newline at end of file diff --git a/README.md b/README.md index c0dcdaf..abf44a0 100644 --- a/README.md +++ b/README.md @@ -1,17 +1,17 @@ -# 📚 KIT Anny Booking Automation +# 📚 Anny Booking Automation -This Python project automates booking of study spaces or resources via the [anny.eu](https://anny.eu) platform used by the KIT (Karlsruhe Institute of Technology) library system. It logs in automatically using KIT SAML SSO, searches for available slots, and makes bookings without user interaction — ideal for recurring reservations. +This Python project automates booking of study spaces or resources via the [anny.eu](https://anny.eu) platform used by university library systems. It logs in automatically using SAML SSO (with pluggable provider support), searches for available slots, and makes bookings without user interaction — ideal for recurring reservations. --- ## ⚙️ Features -- 🔐 Automatic KIT login via SAML SSO -- 📆 Configurable 3-days-ahead reservation window -- 🔎 Auto-detection of available time slots -- ⏳ Pre-login shortly before midnight to instantly book a slot at 00:00 -- 🛠️ Clean and modular object-oriented codebase -- 🔁 Fully automated execution using [cron-job.org](https://cron-job.org) + GitHub API +- 🔐 Pluggable SSO providers (KIT included, easily extendable for other universities) +- 📆 Configurable 3-days-ahead reservation window +- 🔎 Auto-detection of available time slots +- ⏳ Smart midnight wait: only waits if within 10 minutes of midnight, otherwise executes immediately +- 🛠️ Clean and modular object-oriented codebase +- 🔁 Fully automated execution using [cron-job.org](https://cron-job.org) + GitHub API - 📦 Easy to extend and maintain --- @@ -25,13 +25,17 @@ anny_booking/ ├── requirements.txt # Python dependencies │ ├── auth/ -│ └── session.py # AnnySession class (login logic) +│ ├── session.py # AnnySession class (login logic) +│ └── providers/ # SSO provider implementations +│ ├── __init__.py # Provider registry +│ ├── base.py # Abstract base class for providers +│ └── kit.py # KIT (Karlsruhe) SSO provider │ ├── booking/ │ └── client.py # BookingClient class (resource booking) │ ├── config/ -│ └── constants.py # API URLs and shared constants +│ └── constants.py # API URLs, timezone, SSO provider, and shared constants │ ├── utils/ │ └── helpers.py # Utility functions @@ -71,6 +75,46 @@ PASSWORD=your_kit_password > 🔒 Never commit this file to version control! +### 4. Configure SSO provider + +In `config/constants.py`, set your SSO provider: + +```python +SSO_PROVIDER = "kit" # Available: kit (add more in auth/providers/) +``` + +--- + +## 🔌 Adding a New SSO Provider + +To add support for another university (e.g., TUM), create a new file `auth/providers/tum.py`: + +```python +from auth.providers.base import SSOProvider +from utils.helpers import extract_html_value + +class TUMProvider(SSOProvider): + name = "TUM" + domain = "tum.de" + + def authenticate(self) -> str: + # Implement TUM-specific SAML authentication + # Use self.session, self.redirect_response, self.username, self.password + # Return the HTML containing the SAMLResponse + pass +``` + +Then register it in `auth/providers/__init__.py`: + +```python +from auth.providers.tum import TUMProvider + +PROVIDERS: dict[str, type[SSOProvider]] = { + "kit": KITProvider, + "tum": TUMProvider, +} +``` + --- ## ⏱️ Automated Execution via cron-job.org + GitHub API @@ -103,9 +147,9 @@ This ensures your booking script runs exactly when needed — already logged in ### How it works -1. `cron-job.org` triggers the GitHub Actions workflow at **23:58** (Europe/Berlin). -2. The script logs in via KIT SAML SSO and maintains an active session. -3. It waits internally until **00:00**. +1. `cron-job.org` triggers the GitHub Actions workflow at **23:58** (Europe/Berlin). +2. The script logs in via your configured SSO provider and maintains an active session. +3. If within 10 minutes of midnight, it waits until **00:00**. Otherwise, it executes immediately (useful for testing or manual runs). 4. As soon as new booking slots are released, it instantly reserves the first suitable slot. --- @@ -161,7 +205,7 @@ jobs: python main.py ``` -> 💡 Store your KIT credentials securely as [GitHub Secrets](https://docs.github.com/en/actions/security-guides/encrypted-secrets): `USERNAME` and `PASSWORD`. +> 💡 Store your university credentials securely as [GitHub Secrets](https://docs.github.com/en/actions/security-guides/encrypted-secrets): `USERNAME` and `PASSWORD`. --- diff --git a/auth/providers/__init__.py b/auth/providers/__init__.py new file mode 100644 index 0000000..1b10234 --- /dev/null +++ b/auth/providers/__init__.py @@ -0,0 +1,16 @@ +from auth.providers.base import SSOProvider +from auth.providers.kit import KITProvider + +# Registry of available SSO providers +PROVIDERS: dict[str, type[SSOProvider]] = { + "kit": KITProvider, +} + + +def get_provider(name: str) -> type[SSOProvider]: + """Get an SSO provider class by name.""" + provider = PROVIDERS.get(name.lower()) + if not provider: + available = ", ".join(PROVIDERS.keys()) + raise ValueError(f"Unknown SSO provider: {name}. Available: {available}") + return provider diff --git a/auth/providers/base.py b/auth/providers/base.py new file mode 100644 index 0000000..07b1e0e --- /dev/null +++ b/auth/providers/base.py @@ -0,0 +1,35 @@ +from abc import ABC, abstractmethod +import requests + + +class SSOProvider(ABC): + """Base class for SSO authentication providers.""" + + # Override these in subclasses + name: str = "base" + domain: str = "" + + def __init__(self, username: str, password: str): + self.username = username + self.password = password + self.session: requests.Session = None + self.redirect_response: requests.Response = None + self.saml_response_html: str = None + + def set_session(self, session: requests.Session): + """Set the shared session from AnnySession.""" + self.session = session + + def set_redirect_response(self, response: requests.Response): + """Set the redirect response from Anny SSO initiation.""" + self.redirect_response = response + + @abstractmethod + def authenticate(self) -> str: + """ + Perform institution-specific authentication. + + Returns: + The HTML containing the SAML response, or raises an exception on failure. + """ + pass diff --git a/auth/providers/kit.py b/auth/providers/kit.py new file mode 100644 index 0000000..71f06bb --- /dev/null +++ b/auth/providers/kit.py @@ -0,0 +1,36 @@ +import html +from auth.providers.base import SSOProvider +from utils.helpers import extract_html_value + + +class KITProvider(SSOProvider): + """SSO provider for Karlsruhe Institute of Technology (KIT).""" + + name = "KIT" + domain = "kit.edu" + + def authenticate(self) -> str: + self.session.headers.pop('x-requested-with', None) + self.session.headers.pop('x-inertia', None) + self.session.headers.pop('x-inertia-version', None) + + csrf_token = extract_html_value( + self.redirect_response.text, + r'name="csrf_token" value="([^"]+)"' + ) + + response = self.session.post( + 'https://idp.scc.kit.edu/idp/profile/SAML2/Redirect/SSO?execution=e1s1', + data={ + 'csrf_token': csrf_token, + 'j_username': self.username, + 'j_password': self.password, + '_eventId_proceed': '', + 'fudis_web_authn_assertion_input': '', + } + ) + + if "/consume" not in html.unescape(response.text): + raise ValueError("KIT authentication failed - invalid credentials or SSO error") + + return response.text diff --git a/auth/session.py b/auth/session.py index ea154a6..38dd2ea 100644 --- a/auth/session.py +++ b/auth/session.py @@ -1,27 +1,38 @@ import requests import urllib.parse import re -import html from config.constants import AUTH_BASE_URL, ANNY_BASE_URL, DEFAULT_HEADERS from utils.helpers import extract_html_value +from auth.providers import get_provider, SSOProvider + class AnnySession: - def __init__(self, username, password): + def __init__(self, username: str, password: str, provider_name: str = "kit"): self.session = requests.Session() self.username = username self.password = password + # Initialize the SSO provider + provider_class = get_provider(provider_name) + self.provider: SSOProvider = provider_class(username, password) + def login(self): try: self._init_headers() self._sso_login() - self._kit_auth() + self._provider_auth() self._consume_saml() - print("✅ Login successful.") + print(f"✅ Login successful via {self.provider.name}.") return self.session.cookies - except Exception as e: + except requests.RequestException as e: + print(f"[Login Error] Network error: {type(e).__name__}") + return None + except ValueError as e: print(f"[Login Error] {e}") return None + except KeyError as e: + print(f"[Login Error] Missing expected field: {e}") + return None def _init_headers(self): self.session.headers.update({ @@ -45,32 +56,17 @@ class AnnySession: 'x-inertia-version': x_inertia_version }) - r2 = self.session.post(f"{AUTH_BASE_URL}/login/sso", json={"domain": "kit.edu"}) + r2 = self.session.post(f"{AUTH_BASE_URL}/login/sso", json={"domain": self.provider.domain}) redirect_url = r2.headers['x-inertia-location'] - self.redirect_response = self.session.get(redirect_url) + redirect_response = self.session.get(redirect_url) - def _kit_auth(self): - self.session.headers.pop('x-requested-with', None) - self.session.headers.pop('x-inertia', None) - self.session.headers.pop('x-inertia-version', None) + # Pass session and redirect response to provider + self.provider.set_session(self.session) + self.provider.set_redirect_response(redirect_response) - csrf_token = extract_html_value(self.redirect_response.text, r'name="csrf_token" value="([^"]+)"') - - r4 = self.session.post( - 'https://idp.scc.kit.edu/idp/profile/SAML2/Redirect/SSO?execution=e1s1', - data={ - 'csrf_token': csrf_token, - 'j_username': self.username, - 'j_password': self.password, - '_eventId_proceed': '', - 'fudis_web_authn_assertion_input': '', - } - ) - - if "/consume" not in html.unescape(r4.text): - raise Exception("KIT authentication failed") - - self.saml_response_html = r4.text + def _provider_auth(self): + """Delegate authentication to the SSO provider.""" + self.saml_response_html = self.provider.authenticate() def _consume_saml(self): consume_url = extract_html_value(self.saml_response_html, r'form action="([^"]+)"') diff --git a/booking/client.py b/booking/client.py index be185ec..f0dcc7d 100644 --- a/booking/client.py +++ b/booking/client.py @@ -1,6 +1,6 @@ import requests +from requests.exceptions import JSONDecodeError from config.constants import RESOURCE_URL, BOOKING_API_BASE, CHECKOUT_FORM_API, ANNY_BASE_URL, SERVICE_ID -from utils.helpers import extract_html_value class BookingClient: def __init__(self, cookies): @@ -30,7 +30,14 @@ class BookingClient: 'filter[pre_order_ids]': '', 'sort': 'name' }) - resources = response.json().get('data', []) + if not response.ok: + print(f"❌ Failed to fetch resources: HTTP {response.status_code}") + return None + try: + resources = response.json().get('data', []) + except (ValueError, JSONDecodeError): + print(f"❌ Invalid JSON response when fetching resources: {response.text[:200]}") + return None return resources[0]['id'] if resources else None def reserve(self, resource_id, start, end): @@ -50,15 +57,32 @@ class BookingClient: ) if not booking.ok: - print("❌ Slot already taken.") + print(f"❌ Booking failed: HTTP {booking.status_code}") + return False + + try: + data = booking.json().get("data", {}) + except (ValueError, JSONDecodeError): + print(f"❌ Invalid JSON response from booking request: {booking.text[:200]}") return False - data = booking.json().get("data", {}) oid = data.get("id") oat = data.get("attributes", {}).get("access_token") + if not oid or not oat: + print("❌ Missing booking ID or access token in response") + return False + checkout = self.session.get(f"{CHECKOUT_FORM_API}?oid={oid}&oat={oat}&stateless=1") - customer = checkout.json().get("default", {}).get("customer", {}) + if not checkout.ok: + print(f"❌ Checkout form failed: HTTP {checkout.status_code}") + return False + + try: + customer = checkout.json().get("default", {}).get("customer", {}) + except (ValueError, JSONDecodeError): + print(f"❌ Invalid JSON response from checkout form: {checkout.text[:200]}") + return False final = self.session.post( f"{BOOKING_API_BASE}/order?oid={oid}&oat={oat}&stateless=1", diff --git a/config/constants.py b/config/constants.py index ca25dc9..106d817 100644 --- a/config/constants.py +++ b/config/constants.py @@ -4,7 +4,9 @@ BOOKING_API_BASE = "https://b.anny.eu/api/v1" CHECKOUT_FORM_API = "https://b.anny.eu/api/ui/checkout-form" RESOURCE_URL = f"{BOOKING_API_BASE}/resources/1-lehrbuchsammlung-eg-und-1-og/children" SERVICE_ID = "449" -RESSOURCE_ID = "5993" # None Will be set dynamically if None, else use the given ID +RESOURCE_ID = "5960" # Will be set dynamically if None, else use the given ID +TIMEZONE = "Europe/Berlin" +SSO_PROVIDER = "kit" # Available: kit (add more in auth/providers/) DEFAULT_HEADERS = { 'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:140.0) Gecko/20100101 Firefox/140.0', diff --git a/main.py b/main.py index 3acdac9..0d21451 100644 --- a/main.py +++ b/main.py @@ -7,20 +7,20 @@ from auth.session import AnnySession from booking.client import BookingClient from utils.helpers import get_future_datetime import pytz -from config.constants import RESSOURCE_ID +from config.constants import RESOURCE_ID, TIMEZONE, SSO_PROVIDER def main(): load_dotenv('.env', override=True) username = os.getenv("USERNAME") password = os.getenv("PASSWORD") - start_time = datetime.datetime.now(pytz.timezone('Europe/Berlin')) + tz = pytz.timezone(TIMEZONE) if not username or not password: print("❌ Missing USERNAME or PASSWORD in .env") return - session = AnnySession(username, password) + session = AnnySession(username, password, provider_name=SSO_PROVIDER) cookies = session.login() if not cookies: @@ -28,8 +28,17 @@ def main(): booking = BookingClient(cookies) - while start_time.day == datetime.datetime.now(pytz.timezone('Europe/Berlin')).day: - time.sleep(1) + # Only wait for midnight if within 10 minutes, otherwise execute immediately + now = datetime.datetime.now(tz) + midnight = (now + datetime.timedelta(days=1)).replace(hour=0, minute=0, second=0, microsecond=0) + seconds_until_midnight = (midnight - now).total_seconds() + max_wait_seconds = 10 * 60 # 10 minutes + + if 0 < seconds_until_midnight <= max_wait_seconds: + print(f"⏳ Waiting {seconds_until_midnight:.0f} seconds until midnight...") + time.sleep(seconds_until_midnight) + elif seconds_until_midnight > max_wait_seconds: + print(f"⚡ More than 10 min until midnight, executing immediately...") times = [ { @@ -47,19 +56,21 @@ def main(): ] for time_ in times: + try: + start = get_future_datetime(hour=time_['start']) + end = get_future_datetime(hour=time_['end']) - start = get_future_datetime(hour=time_['start']) - end = get_future_datetime(hour=time_['end']) + if RESOURCE_ID: + resource_id = RESOURCE_ID + else: + resource_id = booking.find_available_resource(start, end) - if RESSOURCE_ID: - resource_id = RESSOURCE_ID - else: - resource_id = booking.find_available_resource(start, end) - - if resource_id: - booking.reserve(resource_id, start, end) - else: - print("⚠️ No available slots found.") + if resource_id: + booking.reserve(resource_id, start, end) + else: + print("⚠️ No available slots found.") + except Exception as e: + print(f"❌ Error booking slot {time_['start']}-{time_['end']}: {e}") if __name__ == "__main__": main() diff --git a/utils/helpers.py b/utils/helpers.py index 1e4899f..8821547 100644 --- a/utils/helpers.py +++ b/utils/helpers.py @@ -2,9 +2,10 @@ import re import html import datetime import pytz +from config.constants import TIMEZONE def get_future_datetime(days_ahead=3, hour="13:00:00"): - dt = datetime.datetime.now(pytz.timezone('Europe/Berlin')) + datetime.timedelta(days=days_ahead) + dt = datetime.datetime.now(pytz.timezone(TIMEZONE)) + datetime.timedelta(days=days_ahead) return dt.strftime(f"%Y-%m-%dT{hour}+02:00") def extract_html_value(text, pattern):