From 23c703fec027a1c75d0d3f3989e58ed13f204380 Mon Sep 17 00:00:00 2001 From: Italo Valcy Date: Fri, 10 Jul 2026 13:27:41 -0300 Subject: [PATCH] feat: admin web UI for LTI management commands Add an admin-only web UI (MANAGEMENT sidebar) mirroring the `flask lti ...` commands: list/mint registration tokens, rotate a registration key, purge retired keys and show a registration's public key. The operations are extracted into shared helpers so the CLI and the web routes stay in sync; a small LtiError maps to a click.ClickException (CLI) or a JSON HTTP error (web). Routes are gated with @check_user_category(["admin"]) and the sidebar link only shows when the optional "lti" module is enabled. Co-Authored-By: Claude Opus 4.8 --- apps/lti/keys.py | 21 ++ apps/lti/routes.py | 237 +++++++++--- apps/templates/includes/sidebar.html | 10 + apps/templates/pages/lti_management.html | 435 +++++++++++++++++++++++ tests/test_lti.py | 117 ++++++ 5 files changed, 780 insertions(+), 40 deletions(-) create mode 100644 apps/templates/pages/lti_management.html diff --git a/apps/lti/keys.py b/apps/lti/keys.py index 81ee764..98546b8 100644 --- a/apps/lti/keys.py +++ b/apps/lti/keys.py @@ -7,6 +7,7 @@ cached tokens keep finding the old kid (publish-then-switch rollover). Key file paths are stored in lti_config relative to DATA_DIR. """ +import datetime import hashlib import json import os @@ -92,6 +93,26 @@ def retire_keypair(private_rel, public_rel): os.replace(src, os.path.join(retired_dir(), os.path.basename(rel_path))) +def list_retired_keys(): + """Return retired key files as {name, retired_at} (mtime), newest first, + for the management UI's overview of what is still published in the JWKS + during its grace period.""" + entries = [] + if not os.path.isdir(retired_dir()): + return entries + for fname in sorted(os.listdir(retired_dir())): + path = os.path.join(retired_dir(), fname) + if os.path.isfile(path): + entries.append({ + "name": fname, + "retired_at": datetime.datetime.fromtimestamp( + os.path.getmtime(path), tz=datetime.timezone.utc + ), + }) + entries.sort(key=lambda e: e["retired_at"], reverse=True) + return entries + + def purge_retired_keys(older_than_days): """Delete retired key files older than N days. Returns removed names.""" removed = [] diff --git a/apps/lti/routes.py b/apps/lti/routes.py index d608c60..1c8cfb7 100644 --- a/apps/lti/routes.py +++ b/apps/lti/routes.py @@ -10,8 +10,8 @@ import click import requests from flask import current_app as app -from flask import jsonify, redirect, request, session, url_for -from flask_login import login_user +from flask import jsonify, redirect, render_template, request, session, url_for +from flask_login import current_user, login_required, login_user from pylti1p3.contrib.flask import ( FlaskCacheDataStorage, FlaskMessageLaunch, @@ -19,16 +19,18 @@ FlaskRequest, ) from apps import cache, db -from apps.audit_mixin import get_remote_addr, utcnow +from apps.audit_mixin import check_user_category, get_remote_addr, utcnow from apps.authentication.models import LoginLogging, Users from apps.config import app_config from apps.lti import blueprint from apps.lti.keys import ( JWKS_CACHE_KEY, JWKS_CACHE_TIMEOUT, + abs_key_path, build_jwks, delete_keypair, generate_keypair, + list_retired_keys, purge_retired_keys, retire_keypair, ) @@ -490,13 +492,18 @@ def register(): return REGISTRATION_CLOSE_PAGE -# ----------------------------------------------------------------------- CLI +# ------------------------------------------------------- shared management ops +# The operations below back both the `flask lti ...` CLI commands and the +# admin web UI (/lti/manage); LtiError carries a human-readable reason the +# caller maps to a click.ClickException (CLI) or an HTTP error (web). + +class LtiError(Exception): + """A management operation failed for a reason worth showing the admin.""" + -@blueprint.cli.command("mint-registration-token") -@click.option("--label", required=True, help="Who this token is for (audit)") -@click.option("--ttl-hours", default=24, type=int, help="Token validity in hours") def mint_registration_token(label, ttl_hours): - """Mint a one-time dynamic registration token and print its URL.""" + """Create a one-time dynamic registration token and return its URL. The + plaintext token is only available here (only its hash is stored).""" token = secrets.token_urlsafe(32) db.session.add(LtiRegistrationToken( token_hash=LtiRegistrationToken.hash_token(token), @@ -504,8 +511,179 @@ def mint_registration_token(label, ttl_hours): expires_at=utcnow() + timedelta(hours=ttl_hours), )) db.session.commit() + return f"{lti_base_url()}/register/?token={token}" + + +def _find_registration(issuer, client_id): + """Return (row, registration) for the issuer/client_id or raise LtiError.""" + row = LtiConfig.query.filter_by( + issuer=normalize_issuer(issuer), is_deleted=False + ).first() + if not row: + raise LtiError(f"No lti_config entry for issuer {issuer}") + registration = row.get_registration(client_id) + if not registration: + raise LtiError(f"No registration with client_id {client_id} for issuer {issuer}") + return row, registration + + +def get_registration_public_key(issuer, client_id=None): + """Return a registration's public key PEM (raises LtiError if missing).""" + _, registration = _find_registration(issuer, client_id) + public_key_file = registration.get("public_key_file") + if not public_key_file: + raise LtiError("Registration has no public key file recorded") + with open(abs_key_path(public_key_file)) as f: + return f.read().strip() + + +def rotate_registration_key(issuer, client_id=None): + """Rotate a registration's keypair (publish-then-switch): the new key is + published in /lti/jwks/, the registration switches to it, and the old key + is retired but stays published until purge_retired_keys(). Returns + (issuer, client_id, new_private_rel).""" + row, registration = _find_registration(issuer, client_id) + old_private = registration.get("private_key_file") + old_public = registration.get("public_key_file") + private_rel, public_rel = generate_keypair(row.issuer, registration["client_id"]) + registration["private_key_file"] = private_rel + registration["public_key_file"] = public_rel + row.upsert_registration(registration) + db.session.commit() + retire_keypair(old_private, old_public) + cache.delete(JWKS_CACHE_KEY) + return row.issuer, registration["client_id"], private_rel + + +def purge_retired_keys_op(older_than_days): + """Delete retired key files older than the grace period; returns names.""" + removed = purge_retired_keys(older_than_days) + cache.delete(JWKS_CACHE_KEY) + return removed + + +def list_registrations(): + """Flat list of every active registration across all issuers, for the + management UI (rotate-key/show-public-key act on issuer + client_id).""" + result = [] + for row in LtiConfig.query.filter_by(is_deleted=False).order_by(LtiConfig.issuer).all(): + for reg in row.config: + result.append({ + "issuer": row.issuer, + "client_id": reg.get("client_id"), + "deployment_ids": reg.get("deployment_ids") or [], + "is_default": bool(reg.get("default")), + }) + return result + + +# ----------------------------------------------------------------- admin web UI + +@blueprint.route("/manage", methods=["GET"]) +@login_required +@check_user_category(["admin"]) +def manage(): + """LTI management dashboard (MANAGEMENT sidebar): the web equivalent of + the `flask lti ...` commands - registrations (rotate key / show public + key), one-time registration tokens (mint / list) and retired keys + (purge).""" + return render_template( + "pages/lti_management.html", + registrations=list_registrations(), + tokens=LtiRegistrationToken.query.order_by( + LtiRegistrationToken.id.desc() + ).all(), + retired_keys=list_retired_keys(), + ) + + +@blueprint.route("/manage/mint-token", methods=["POST"]) +@login_required +@check_user_category(["admin"]) +def manage_mint_token(): + label = (request.form.get("label") or "").strip() + if not label: + return jsonify({"error": "A label is required (who the token is for)"}), 400 + try: + ttl_hours = int(request.form.get("ttl_hours") or 24) + except ValueError: + return jsonify({"error": "TTL must be a whole number of hours"}), 400 + if ttl_hours <= 0: + return jsonify({"error": "TTL must be a positive number of hours"}), 400 + url = mint_registration_token(label, ttl_hours) + app.logger.info( + f"LTI registration token minted by {current_user.username} " + f"label={label!r} ttl_hours={ttl_hours}" + ) + return jsonify({"url": url, "label": label, "ttl_hours": ttl_hours}) + + +@blueprint.route("/manage/public-key", methods=["GET"]) +@login_required +@check_user_category(["admin"]) +def manage_public_key(): + issuer = request.args.get("issuer", "") + client_id = request.args.get("client_id") or None + try: + pem = get_registration_public_key(issuer, client_id) + except LtiError as exc: + return jsonify({"error": str(exc)}), 404 + return jsonify({"issuer": issuer, "client_id": client_id, "public_key": pem}) + + +@blueprint.route("/manage/rotate-key", methods=["POST"]) +@login_required +@check_user_category(["admin"]) +def manage_rotate_key(): + issuer = request.form.get("issuer", "") + client_id = request.form.get("client_id") or None + try: + issuer, client_id, _ = rotate_registration_key(issuer, client_id) + except LtiError as exc: + return jsonify({"error": str(exc)}), 404 + app.logger.info( + f"LTI key rotated by {current_user.username} issuer={issuer} client_id={client_id}" + ) + return jsonify({ + "result": ( + f"Rotated key for {client_id} @ {issuer}. The old key is retired " + "but stays published in /lti/jwks/ until you purge it after the " + "grace period." + ), + }) + + +@blueprint.route("/manage/purge-retired-keys", methods=["POST"]) +@login_required +@check_user_category(["admin"]) +def manage_purge_retired_keys(): + try: + older_than_days = int(request.form.get("older_than_days") or 30) + except ValueError: + return jsonify({"error": "Grace period must be a whole number of days"}), 400 + if older_than_days < 0: + return jsonify({"error": "Grace period cannot be negative"}), 400 + removed = purge_retired_keys_op(older_than_days) + app.logger.info( + f"LTI retired keys purged by {current_user.username} " + f"older_than_days={older_than_days} removed={len(removed)}" + ) + return jsonify({ + "result": f"Removed {len(removed)} retired key file(s)", + "removed": removed, + }) + + +# ----------------------------------------------------------------------- CLI + +@blueprint.cli.command("mint-registration-token") +@click.option("--label", required=True, help="Who this token is for (audit)") +@click.option("--ttl-hours", default=24, type=int, help="Token validity in hours") +def mint_registration_token_cmd(label, ttl_hours): + """Mint a one-time dynamic registration token and print its URL.""" + url = mint_registration_token(label, ttl_hours) click.echo("Registration URL (shown once, hand it to the LMS admin):") - click.echo(f"{lti_base_url()}/register/?token={token}") + click.echo(url) @blueprint.cli.command("list-registration-tokens") @@ -527,18 +705,10 @@ def show_public_key(issuer, client_id): Moodle's symptom is a token.php 404 with 'jwks_helper::fix_jwks_alg(): ... null given' on the first grade/roster call (launches still work, they never need platform->tool calls).""" - row = LtiConfig.query.filter_by(issuer=normalize_issuer(issuer), is_deleted=False).first() - if not row: - raise click.ClickException(f"No lti_config entry for issuer {issuer}") - registration = row.get_registration(client_id) - if not registration: - raise click.ClickException(f"No registration with client_id {client_id} for issuer {issuer}") - public_key_file = registration.get("public_key_file") - if not public_key_file: - raise click.ClickException("Registration has no public key file recorded") - from apps.lti.keys import abs_key_path - with open(abs_key_path(public_key_file)) as f: - click.echo(f.read().strip()) + try: + click.echo(get_registration_public_key(issuer, client_id)) + except LtiError as exc: + raise click.ClickException(str(exc)) @blueprint.cli.command("rotate-key") @@ -548,24 +718,12 @@ def rotate_key(issuer, client_id): """Rotate a registration's keypair (publish-then-switch): the new key is published in /lti/jwks/, the registration switches to it, and the old key is retired but stays published until purge-retired-keys.""" - row = LtiConfig.query.filter_by(issuer=normalize_issuer(issuer), is_deleted=False).first() - if not row: - raise click.ClickException(f"No lti_config entry for issuer {issuer}") - registration = row.get_registration(client_id) - if not registration: - raise click.ClickException(f"No registration with client_id {client_id} for issuer {issuer}") - - old_private = registration.get("private_key_file") - old_public = registration.get("public_key_file") - private_rel, public_rel = generate_keypair(row.issuer, registration["client_id"]) - registration["private_key_file"] = private_rel - registration["public_key_file"] = public_rel - row.upsert_registration(registration) - db.session.commit() - retire_keypair(old_private, old_public) - cache.delete(JWKS_CACHE_KEY) + try: + issuer, client_id, private_rel = rotate_registration_key(issuer, client_id) + except LtiError as exc: + raise click.ClickException(str(exc)) click.echo( - f"Rotated key for issuer={row.issuer} client_id={registration['client_id']}: " + f"Rotated key for issuer={issuer} client_id={client_id}: " f"new key {private_rel}, old key retired (still published in /lti/jwks/; " f"run 'flask lti purge-retired-keys' after the grace period)" ) @@ -575,8 +733,7 @@ def rotate_key(issuer, client_id): @click.option("--older-than-days", default=30, type=int, help="Grace period in days") def purge_retired_keys_cmd(older_than_days): """Delete retired key files older than the grace period.""" - removed = purge_retired_keys(older_than_days) - cache.delete(JWKS_CACHE_KEY) + removed = purge_retired_keys_op(older_than_days) click.echo(f"Removed {len(removed)} retired key file(s)") for fname in removed: click.echo(f" {fname}") diff --git a/apps/templates/includes/sidebar.html b/apps/templates/includes/sidebar.html index ba061d0..88195ff 100644 --- a/apps/templates/includes/sidebar.html +++ b/apps/templates/includes/sidebar.html @@ -147,6 +147,16 @@

+ {% if config.get('ENABLE_LTI') and current_user.category == "admin" %} +
  • + + +

    + LTI +

    +     +
    • + {% endif %}
  • diff --git a/apps/templates/pages/lti_management.html b/apps/templates/pages/lti_management.html new file mode 100644 index 0000000..6a71312 --- /dev/null +++ b/apps/templates/pages/lti_management.html @@ -0,0 +1,435 @@ +{% extends "layouts/base.html" %} + +{% block title %} LTI Management {% endblock title %} + + +{% block body_class %} sidebar-mini layout-navbar-fixed {% endblock body_class %} + + +{% block stylesheets %} + + + + + + + + + + +{% endblock stylesheets %} + +{% block content %} + + +
    + +
    +
    +
    +
    +

    LTI Management

    +
    +
    +
      +
    1. Home
      2. +
    2. LTI
      3. +
    +
    +
    +
    +
    + + +
    + + +
    +
    +
    +
    +

    Platform Registrations

    +
    +
    +

    + LTI 1.3 platform registrations (created by dynamic registration). + Rotate a registration's signing key or show its public key (PEM) to + paste into the LMS when it cannot fetch /lti/jwks/. +

    + + + + + + + + + + + + {% for reg in registrations %} + + + + + + + + {% endfor %} + +
    IssuerClient IDDeployment IDsDefaultActions
    {{ reg.issuer }}{{ reg.client_id }}{{ reg.deployment_ids | join(', ') }}{% if reg.is_default %}default{% endif %} + + +
    +
    +
    +
    +
    + + +
    +
    +
    +
    +

    Registration Tokens

    +
    + +
    +
    +
    +

    + One-time credentials for the dynamic registration endpoint. The token + itself is shown only once, at mint time; only its hash is stored. +

    + + + + + + + + + + + + + {% for token in tokens %} + + + + + + + + + {% endfor %} + +
    IDLabelCreatedExpiresUsedStatus
    {{ token.id }}{{ token.label }}{{ token.created_at.strftime('%Y-%m-%d %H:%M') if token.created_at else '-' }}{{ token.expires_at.strftime('%Y-%m-%d %H:%M') if token.expires_at else '-' }}{{ token.used_at.strftime('%Y-%m-%d %H:%M') if token.used_at else '-' }} + {% if token.used_at %} + used + {% elif token.is_expired %} + expired + {% else %} + valid + {% endif %} +
    +
    +
    +
    +
    + + +
    +
    +
    +
    +

    Retired Keys

    +
    + +
    +
    +
    +

    + Rotated keys stay published in /lti/jwks/ during a grace + period so platforms validating cached tokens keep finding the old key. + Purge them once the grace period has passed. +

    + + + + + + + + + {% for key in retired_keys %} + + + + + {% endfor %} + +
    Key FileRetired At (UTC)
    {{ key.name }}{{ key.retired_at.strftime('%Y-%m-%d %H:%M') }}
    +
    +
    +
    +
    + + + + + + + + + + + + + +
    + +
    + + +{% endblock content %} + + +{% block javascripts %} + + + + + + + + + + + + + + + +{% endblock javascripts %} diff --git a/tests/test_lti.py b/tests/test_lti.py index 3ba5210..95b05be 100644 --- a/tests/test_lti.py +++ b/tests/test_lti.py @@ -675,3 +675,120 @@ def test_cookie_check_page_then_redirect_to_platform(self, client): assert location.startswith(ISSUER + "/mod/lti/auth.php") assert "state=" in location and "nonce=" in location assert "client_id=client-1" in location + + +# --- admin management web UI -------------------------------------------------- + +def _ensure_user(username, category): + """Create (once) a password-login user for the web UI tests.""" + user = Users.query.filter_by(username=username).first() + if not user: + user = Users( + username=username, password="lti-web-pass", + email=f"{username}@lti-web.example", category=category, + ) + db.session.add(user) + db.session.commit() + return user + + +def _web_login(client, username): + return client.post( + "/login/", + data={"identifier": username, "password": "lti-web-pass", "login": "1"}, + ) + + +class TestManagementWebUI: + def test_non_admin_is_rejected(self, client): + _ensure_user("ltiwebstudent", "student") + _web_login(client, "ltiwebstudent") + resp = client.get("/lti/manage") + assert b"Unauthorized request" in resp.data + client.get("/logout") + + def test_admin_dashboard_lists_registrations(self, client): + _seed_registration() + _ensure_user("ltiwebadmin", "admin") + _web_login(client, "ltiwebadmin") + resp = client.get("/lti/manage") + assert resp.status_code == 200 + assert b"LTI Management" in resp.data + assert ISSUER.encode() in resp.data + assert b"client-1" in resp.data + + def test_mint_token_creates_row_and_returns_url(self, client): + _web_login(client, "ltiwebadmin") + before = LtiRegistrationToken.query.count() + resp = client.post("/lti/manage/mint-token", data={ + "label": "web-ui token", "ttl_hours": "12", + }) + assert resp.status_code == 200 + data = resp.get_json() + assert "/lti/register/?token=" in data["url"] + assert LtiRegistrationToken.query.count() == before + 1 + row = LtiRegistrationToken.query.order_by( + LtiRegistrationToken.id.desc()).first() + assert row.label == "web-ui token" + # only the hash is stored, never the plaintext token + token_value = data["url"].split("token=", 1)[1] + assert row.token_hash == LtiRegistrationToken.hash_token(token_value) + + def test_mint_token_requires_label(self, client): + _web_login(client, "ltiwebadmin") + resp = client.post("/lti/manage/mint-token", data={"label": " "}) + assert resp.status_code == 400 + assert "label" in resp.get_json()["error"].lower() + + def test_show_public_key(self, client): + _web_login(client, "ltiwebadmin") + resp = client.get("/lti/manage/public-key", query_string={ + "issuer": ISSUER, "client_id": "client-1", + }) + assert resp.status_code == 200 + assert "BEGIN PUBLIC KEY" in resp.get_json()["public_key"] + # unknown issuer -> 404 with a reason + resp = client.get("/lti/manage/public-key", query_string={ + "issuer": "https://nope.example", + }) + assert resp.status_code == 404 + assert "No lti_config entry" in resp.get_json()["error"] + + def test_rotate_key_publish_then_switch(self, client): + _web_login(client, "ltiwebadmin") + row = LtiConfig.query.filter_by(issuer=ISSUER).first() + old_public = row.get_registration("client-1")["public_key_file"] + resp = client.post("/lti/manage/rotate-key", data={ + "issuer": ISSUER, "client_id": "client-1", + }) + assert resp.status_code == 200 + assert "Rotated key" in resp.get_json()["result"] + db.session.expire_all() + row = LtiConfig.query.filter_by(issuer=ISSUER).first() + assert row.get_registration("client-1")["public_key_file"] != old_public + assert not os.path.exists(lti_keys.abs_key_path(old_public)) + assert os.path.basename(old_public) in os.listdir(lti_keys.retired_dir()) + + def test_rotate_key_unknown_issuer(self, client): + _web_login(client, "ltiwebadmin") + resp = client.post("/lti/manage/rotate-key", data={ + "issuer": "https://nope.example", + }) + assert resp.status_code == 404 + assert "No lti_config entry" in resp.get_json()["error"] + + def test_purge_retired_keys(self, client): + _web_login(client, "ltiwebadmin") + # the rotate test above left a retired key behind: the dashboard must + # render it (exercises the retired-key row / timestamp formatting) + assert os.listdir(lti_keys.retired_dir()) != [] + resp = client.get("/lti/manage") + assert resp.status_code == 200 + assert b"Retired Keys" in resp.data + resp = client.post("/lti/manage/purge-retired-keys", data={ + "older_than_days": "0", + }) + assert resp.status_code == 200 + assert "Removed" in resp.get_json()["result"] + assert os.listdir(lti_keys.retired_dir()) == [] + client.get("/logout")