forked from Ugric/cov-to-ics
439 lines
9.8 KiB
Python
439 lines
9.8 KiB
Python
from bs4 import BeautifulSoup
|
|
from datetime import datetime, timezone
|
|
import pytz
|
|
import hashlib
|
|
import os
|
|
from flask import (
|
|
Flask,
|
|
request,
|
|
send_file,
|
|
render_template,
|
|
Response,
|
|
redirect,
|
|
url_for,
|
|
session,
|
|
)
|
|
import requests
|
|
from requests_ntlm import HttpNtlmAuth
|
|
import urllib
|
|
import html
|
|
from cov2ics import *
|
|
import base64
|
|
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
|
|
from dotenv import load_dotenv
|
|
import json
|
|
from flask_wtf.csrf import CSRFProtect
|
|
from flask_wtf.csrf import CSRFProtect
|
|
|
|
|
|
load_dotenv()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Configuration
|
|
# ---------------------------------------------------------------------------
|
|
|
|
app = Flask(__name__)
|
|
|
|
app.config["SECRET_KEY"] = os.getenv("SECRET_KEY")
|
|
|
|
app.config.update(
|
|
SESSION_COOKIE_SECURE=True,
|
|
SESSION_COOKIE_HTTPONLY=True,
|
|
SESSION_COOKIE_SAMESITE="Lax",
|
|
)
|
|
|
|
csrf = CSRFProtect(app)
|
|
|
|
|
|
# AES_KEY should be a 64-character hexadecimal string representing
|
|
# exactly 32 random bytes (256 bits).
|
|
#
|
|
# Generate one with:
|
|
#
|
|
# python -c "import secrets; print(secrets.token_hex(32))"
|
|
#
|
|
# Store the result in your environment as AES_KEY.
|
|
aes_key = bytes.fromhex(os.environ["AES_KEY"])
|
|
|
|
if len(aes_key) != 32:
|
|
raise RuntimeError("AES_KEY must be exactly 32 bytes / 64 hex characters")
|
|
|
|
aes = AESGCM(aes_key)
|
|
|
|
|
|
URL = os.getenv("URL")
|
|
WEBCAL_URL = os.getenv("WEBCAL_URL")
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Stateless authentication token
|
|
# ---------------------------------------------------------------------------
|
|
|
|
AUTH_TOKEN_VERSION = b"\x01"
|
|
AUTH_TOKEN_AAD = b"timtabla-auth-v1"
|
|
|
|
# 12 bytes is the recommended nonce size for AES-GCM.
|
|
NONCE_SIZE = 12
|
|
|
|
# Ten years for the browser login cookie.
|
|
AUTH_MAX_AGE = 10 * 365 * 24 * 60 * 60
|
|
|
|
|
|
def b64encode(data: bytes) -> str:
|
|
"""Base64URL encode without '=' padding."""
|
|
return base64.urlsafe_b64encode(data).rstrip(b"=").decode("ascii")
|
|
|
|
|
|
def b64decode(data: str) -> bytes:
|
|
"""Base64URL decode with optional '=' padding."""
|
|
padding = "=" * (-len(data) % 4)
|
|
return base64.urlsafe_b64decode(data + padding)
|
|
|
|
|
|
def generate_auth_token(username: str, password: str) -> str:
|
|
"""
|
|
Create a stateless encrypted authentication token.
|
|
|
|
Token layout:
|
|
|
|
version (1 byte)
|
|
nonce (12 bytes)
|
|
ciphertext + GCM authentication tag
|
|
|
|
The plaintext contains the university credentials.
|
|
"""
|
|
|
|
payload = {
|
|
"u": username,
|
|
"p": password,
|
|
"exp": int(datetime.now(timezone.utc).timestamp()) + AUTH_MAX_AGE,
|
|
}
|
|
|
|
plaintext = json.dumps(
|
|
payload,
|
|
separators=(",", ":"),
|
|
).encode("utf-8")
|
|
|
|
nonce = os.urandom(NONCE_SIZE)
|
|
|
|
ciphertext = aes.encrypt(
|
|
nonce,
|
|
plaintext,
|
|
AUTH_TOKEN_AAD,
|
|
)
|
|
|
|
token = AUTH_TOKEN_VERSION + nonce + ciphertext
|
|
|
|
return b64encode(token)
|
|
|
|
|
|
def decrypt_auth_token(token: str):
|
|
"""
|
|
Decrypt and validate an authentication token.
|
|
|
|
Returns:
|
|
(username, password)
|
|
|
|
Returns:
|
|
None
|
|
|
|
if the token is invalid, expired, or has been tampered with.
|
|
"""
|
|
|
|
try:
|
|
raw = b64decode(token)
|
|
|
|
# Minimum:
|
|
# 1 byte version
|
|
# 12 byte nonce
|
|
# 16 byte GCM authentication tag
|
|
if len(raw) < 1 + NONCE_SIZE + 16:
|
|
return None
|
|
|
|
version = raw[0:1]
|
|
|
|
if version != AUTH_TOKEN_VERSION:
|
|
return None
|
|
|
|
nonce = raw[1 : 1 + NONCE_SIZE]
|
|
ciphertext = raw[1 + NONCE_SIZE :]
|
|
|
|
plaintext = aes.decrypt(
|
|
nonce,
|
|
ciphertext,
|
|
AUTH_TOKEN_AAD,
|
|
)
|
|
|
|
payload = json.loads(plaintext.decode("utf-8"))
|
|
|
|
username = payload["u"]
|
|
password = payload["p"]
|
|
expiry = int(payload["exp"])
|
|
|
|
if not isinstance(username, str) or not isinstance(password, str):
|
|
return None
|
|
|
|
if datetime.now(timezone.utc).timestamp() >= expiry:
|
|
return None
|
|
|
|
return username, password
|
|
|
|
except Exception:
|
|
# This deliberately catches malformed Base64, invalid JSON,
|
|
# invalid GCM tags, missing fields, etc.
|
|
return None
|
|
|
|
|
|
def get_browser_auth():
|
|
"""
|
|
Get credentials from the browser's encrypted auth cookie.
|
|
|
|
Returns:
|
|
(status, username, password)
|
|
"""
|
|
|
|
token = request.cookies.get("auth")
|
|
|
|
if not token:
|
|
return False, None, None
|
|
|
|
credentials = decrypt_auth_token(token)
|
|
|
|
if credentials is None:
|
|
return False, None, None
|
|
|
|
username, password = credentials
|
|
|
|
try:
|
|
status = fetch_auth_status(
|
|
f"COVENTRY\\{username}",
|
|
password,
|
|
)
|
|
except Exception:
|
|
status = False
|
|
|
|
return status, username, password
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Routes
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@app.route("/")
|
|
def index():
|
|
status, username, password = get_browser_auth()
|
|
|
|
if status:
|
|
return redirect("/dashboard")
|
|
|
|
return render_template("home.jinja")
|
|
|
|
|
|
@app.route("/about")
|
|
def about():
|
|
return render_template("about.jinja")
|
|
|
|
|
|
@app.route("/login", methods=["GET"])
|
|
def login():
|
|
session.clear()
|
|
|
|
status, username, password = get_browser_auth()
|
|
|
|
if status:
|
|
return redirect("/dashboard")
|
|
|
|
return render_template("login.jinja")
|
|
|
|
|
|
@app.route("/login", methods=["POST"])
|
|
def login_post():
|
|
username = request.form.get("username", "").strip()
|
|
password = request.form.get("password", "")
|
|
|
|
if not username or not password:
|
|
return (
|
|
render_template(
|
|
"login.jinja",
|
|
error="Username and password are required.",
|
|
),
|
|
400,
|
|
)
|
|
|
|
# Verify the credentials before storing them in the encrypted cookie.
|
|
try:
|
|
status = fetch_auth_status(
|
|
f"COVENTRY\\{username}",
|
|
password,
|
|
)
|
|
except Exception:
|
|
status = False
|
|
|
|
if not status:
|
|
return (
|
|
render_template(
|
|
"login.jinja",
|
|
error="Invalid university username or password.",
|
|
),
|
|
401,
|
|
)
|
|
|
|
token = generate_auth_token(username, password)
|
|
|
|
resp = redirect("/dashboard")
|
|
|
|
resp.set_cookie(
|
|
"auth",
|
|
token,
|
|
max_age=AUTH_MAX_AGE,
|
|
secure=True,
|
|
httponly=True,
|
|
samesite="Lax",
|
|
)
|
|
|
|
return resp
|
|
|
|
|
|
@app.route("/logout", methods=["POST"])
|
|
def logout():
|
|
session.clear()
|
|
|
|
resp = redirect("/")
|
|
|
|
resp.delete_cookie(
|
|
"auth",
|
|
secure=True,
|
|
httponly=True,
|
|
samesite="Lax",
|
|
)
|
|
|
|
return resp
|
|
|
|
|
|
@app.route("/dashboard")
|
|
def dashboard():
|
|
status, username, password = get_browser_auth()
|
|
|
|
if not status:
|
|
return redirect("/login")
|
|
|
|
# Generate a completely new token for the feed URL.
|
|
#
|
|
# This means the browser cookie and the calendar feed do not
|
|
# contain identical ciphertext.
|
|
token = generate_auth_token(username, password)
|
|
|
|
webcal_url = WEBCAL_URL+"/timtabla.ics?" + urllib.parse.urlencode({"t": token})
|
|
|
|
return render_template(
|
|
"dashboard.jinja",
|
|
username=username,
|
|
webcal_url=webcal_url,
|
|
)
|
|
|
|
|
|
@app.route("/download-ics", methods=["GET"])
|
|
def download_ics():
|
|
status, username, password = get_browser_auth()
|
|
|
|
if not status:
|
|
return redirect("/login")
|
|
|
|
try:
|
|
data, start_year, start_month, end_year, end_month = build_calendar(
|
|
fetch_timetable(
|
|
f"COVENTRY\\{username}",
|
|
password,
|
|
)
|
|
)
|
|
|
|
return Response(
|
|
data,
|
|
mimetype="text/calendar",
|
|
headers={
|
|
"Content-Disposition": (
|
|
f"attachment; "
|
|
f'filename="{username}_timtabla_'
|
|
f"{start_month}-{start_year}_to_"
|
|
f'{end_month}-{end_year}.ics"'
|
|
)
|
|
},
|
|
)
|
|
|
|
except Exception:
|
|
return "Calendar Not Found", 404
|
|
|
|
|
|
@app.route("/timtabla.ics", methods=["GET"])
|
|
def live_ics():
|
|
"""
|
|
Live calendar endpoint.
|
|
|
|
Example:
|
|
|
|
/timtabla.ics?t=BASE64URL_TOKEN
|
|
|
|
The token is a bearer credential. Anyone possessing the URL
|
|
can access the timetable until the token expires or the
|
|
university password stops working.
|
|
"""
|
|
|
|
token = request.args.get("t")
|
|
|
|
if not token:
|
|
return "Invalid calendar URL", 401
|
|
|
|
credentials = decrypt_auth_token(token)
|
|
|
|
if credentials is None:
|
|
return "Invalid or expired calendar URL", 401
|
|
|
|
username, password = credentials
|
|
|
|
try:
|
|
# Optionally verify that the university credentials still work.
|
|
status = fetch_auth_status(
|
|
f"COVENTRY\\{username}",
|
|
password,
|
|
)
|
|
|
|
if not status:
|
|
return "University authentication failed", 401
|
|
|
|
data, start_year, start_month, end_year, end_month = build_calendar(
|
|
fetch_timetable(
|
|
f"COVENTRY\\{username}",
|
|
password,
|
|
)
|
|
)
|
|
|
|
response = Response(
|
|
data,
|
|
mimetype="text/calendar",
|
|
headers={
|
|
"Content-Disposition": (
|
|
f'attachment; filename="{username}_timtabla.ics"'
|
|
),
|
|
# Don't allow intermediary caches to retain the
|
|
# generated timetable.
|
|
"Cache-Control": "no-store",
|
|
# Prevent the private feed URL from being sent as
|
|
# a Referer header to another site.
|
|
"Referrer-Policy": "no-referrer",
|
|
},
|
|
)
|
|
|
|
return response
|
|
|
|
except Exception:
|
|
return "Calendar Not Found", 404
|
|
|
|
|
|
if __name__ == "__main__":
|
|
app.run(
|
|
debug=True,
|
|
host="0.0.0.0",
|
|
port=3000,
|
|
)
|