diff --git a/api/ADMIN_NOTIFICATIONS.md b/api/ADMIN_NOTIFICATIONS.md new file mode 100644 index 0000000..b0db83a --- /dev/null +++ b/api/ADMIN_NOTIFICATIONS.md @@ -0,0 +1,110 @@ +# Admin notifications and role requests + +## What administrators are told about + +| Event | When | Where | +|---|---|---| +| New user | A user signs in for the first time (including a user an admin invited earlier) | Teams + email to admins | +| Equipment awaiting approval | A non-admin registers equipment, or submits changes that need review (one notification each time it enters review; resubmitting while it is still pending does not notify again) | Teams + email to admins | +| Role request | A user asks for a different role from their Settings page | Teams + email to admins | +| Role decision | An admin approves or rejects a role request | Email to the requester (only if email is configured; they always see it on their Settings page) | + +Each notification links to the page where it is handled (`/admin/users`, +`/admin/reviews`, `/admin/role-requests`, `/settings`) when +`PLATFORM_BASE_URL` is set. + +## How it works + +Every event writes a row to the `platform_notifications` outbox in the same +database transaction as the event, so a notification exists exactly when the +event does, and a unique key stops duplicates (the sign-in check runs on every +page load but notifies once per user). A background pass delivers new rows +straight away, and each API worker runs a sweeper every +`NOTIFICATION_SWEEP_SECONDS` (default 60) for anything missed. Each +row is claimed on its own just before sending, under a lease and a claim +token, so the two API workers never send the same row at once. Failed +deliveries are retried with backoff (1, 2, 4 … minutes, at most an hour +apart) up to `NOTIFICATION_MAX_ATTEMPTS` (default 10), then marked failed; if +an SMTP server refuses only some recipients, only those are retried. Delivery +is at-least-once: a crash between sending and recording can send twice. A +channel that is not configured is skipped, not queued: configuring email later +does not send old notifications. Delivered rows are deleted after +`NOTIFICATION_RETENTION_DAYS` (default 90). Stored and logged errors are short +categories (for example `HTTP 404` or `SMTPRecipientsRefused`) and never +contain webhook URLs or credentials. + +Request handlers can only insert into the outbox; only the background worker +role (`api_worker`) reads and updates it. + +## Configuration + +Set these in `dt-api-secret` (see `geddes/k8s/01-secrets.yaml.example`) and +restart the API. Teams alone is a supported setup: with no email transport +configured, email is skipped for every notification. `GET /api/health` reports +`runtime.admin_teams_notifications_configured` and +`runtime.email_notifications_configured`. + +### Teams + +`ADMIN_TEAMS_WEBHOOK_URL`: the HTTP trigger URL of a Power Automate flow (or +Logic App), the same kind used for the retrain alerts: + +1. Power Automate → Create → Instant cloud flow → trigger **When a Teams + webhook request is received** (or **When a HTTP request is received**). +2. Add **Post card in a chat or channel**; post as Flow bot to the channel or + chat you want, and set the card to the trigger body (`triggerBody()`). The + API sends an Adaptive Card. +3. Save and copy the trigger URL into `ADMIN_TEAMS_WEBHOOK_URL`. + +### Email + +Recipients are `ADMIN_NOTIFICATION_EMAILS` (comma separated) or, if unset, +every active admin's email address in the platform. Choose one transport: + +- **SMTP relay:** `SMTP_HOST`, `SMTP_PORT` (default 587), `SMTP_SECURITY` + (`starttls` default, `ssl` or `none`; any other value is refused rather + than falling back to an unencrypted connection), `SMTP_FROM`, and `SMTP_USERNAME` / + `SMTP_PASSWORD` if the relay requires login. Use a relay that accepts mail + from the Geddes cluster. +- **Power Automate:** `EMAIL_WEBHOOK_URL`, a flow triggered by **When a HTTP + request is received** with the schema below, followed by **Send an email + (V2)** using `to` (joined with `;`), `subject` and `html` as the body. It + sends from the account that owns the flow. + + ```json + {"type": "object", "properties": { + "to": {"type": "array", "items": {"type": "string"}}, + "subject": {"type": "string"}, + "text": {"type": "string"}, + "html": {"type": "string"}}} + ``` + +If both are set, SMTP is used. + +## Role requests + +Everyone joins as a Researcher. On **Settings → Your role** a user picks the +role they need and explains why; they can have one pending request, can +cancel it, and can send at most three requests a day. Administrators decide on **Admin → Role Requests**, optionally with +a note. Approval applies the same rules as changing a role on the Users page +(an admin cannot demote themselves or the last admin) and is recorded in the +role audit. Changing the role of an invited user who has not signed in yet +keeps them "invited", so their first sign-in still notifies the admins. The requester sees the decision on their Settings page (and receives an +email when an email transport is configured), and the web app picks up the +new role on their next page load. + +Row-level security on `role_change_requests` (always enforced) lets users see +and cancel only their own requests; only admins see all requests and decide +them. When duplicate accounts are merged, their requests move to the +remaining account and at most one stays pending. + +The sign-in identity sync (`GET /api/admin/users/{email}/role`) is +server-to-server only: the web proxy refuses it and the API rejects calls +that carry a user identity, so users cannot create or merge other people's +accounts or fake sign-in notifications. + +## Deploying + +The new tables are created by the schema migrations: run the +`dt-db-migrate` Job (`geddes/k8s/09-db-migrate-job.yaml`) with the new image +before rolling out the API. diff --git a/api/db_migrations.py b/api/db_migrations.py index 9f190ab..4c9fac5 100644 --- a/api/db_migrations.py +++ b/api/db_migrations.py @@ -44,7 +44,7 @@ # Reported by /api/health. -RLS_REPORTED_TABLES = ("projects", "project_members") + RLS_ENFORCED_TABLES +RLS_REPORTED_TABLES = ("projects", "project_members", "role_change_requests") + RLS_ENFORCED_TABLES def rls_enforcement_mode() -> str: @@ -190,6 +190,8 @@ def run_schema_migrations() -> None: ensure_user_role_audit_table_pg, ) from ml_response_cache import ensure_ml_endpoint_cache_table_pg + from platform_notifications import ensure_platform_notifications_table_pg + from role_requests import ensure_role_change_requests_table_pg lock_conn = get_pg_migration_connection() try: @@ -220,6 +222,8 @@ def run_schema_migrations() -> None: ensure_data_uploads_kind_column_pg() ensure_data_uploads_processing_started_at_pg() ensure_ml_endpoint_cache_table_pg() + ensure_platform_notifications_table_pg() + ensure_role_change_requests_table_pg() # Last: the policies reference tables and columns created above. ensure_row_level_security_pg() finally: diff --git a/api/main.py b/api/main.py index 5e29165..75622c1 100644 --- a/api/main.py +++ b/api/main.py @@ -87,6 +87,10 @@ def _runtime_informational_checks() -> dict[str, bool]: return { "postgres_superuser_password_configured": _env_present("PG_SUPER_PASS"), "postgres_worker_role_configured": _env_present("PG_WORKER_PASS"), + "admin_teams_notifications_configured": _env_present("ADMIN_TEAMS_WEBHOOK_URL"), + "email_notifications_configured": ( + _env_present("SMTP_HOST") or _env_present("EMAIL_WEBHOOK_URL") + ), } @@ -177,14 +181,16 @@ def _run_startup_migrations() -> None: def _start_background_jobs() -> None: - """Recover interrupted background work and start the proposal job sweeper.""" + """Recover interrupted background work and start the background sweepers.""" from metadata_pg import recover_stale_db3_uploads_pg, start_proposal_job_sweeper + from platform_notifications import start_notification_sweeper try: recover_stale_db3_uploads_pg() except Exception: logger.warning("Could not recover stale DB3 uploads at startup", exc_info=True) start_proposal_job_sweeper() + start_notification_sweeper() # ── Startup / shutdown ─────────────────────────────────────────────────────── diff --git a/api/metadata_pg.py b/api/metadata_pg.py index 82ae9a6..c1bcf06 100644 --- a/api/metadata_pg.py +++ b/api/metadata_pg.py @@ -1445,6 +1445,42 @@ def get_equipment_pg(domain_id: str) -> dict[str, Any] | None: conn.close() +def _enqueue_equipment_review_notification( + cur, + *, + registration_id: str, + domain_id: str, + equipment_name: str, + user: PlatformUser, + is_change: bool, +) -> None: + """Tell administrators an equipment registration is waiting for review.""" + from platform_notifications import enqueue_notification + + action = "changes to" if is_change else "a new registration for" + enqueue_notification( + cur, + kind="equipment_review_requested", + # Called once per review cycle (see the callers); registrations are + # upserted, so the id alone would repeat across cycles. + dedupe_key=f"equipment_review_requested:{registration_id}:{uuid.uuid4().hex}", + subject=f"Equipment awaiting approval: {equipment_name}", + message=( + f"{user.name or user.id} submitted {action} {equipment_name}. " + "It needs an administrator's review before it can be used." + ), + facts=[ + ("Equipment", equipment_name), + ("Equipment ID", domain_id), + ("Submitted by", user.name or user.id), + ("Email", user.email), + ("Organization", user.organization), + ("Type", "Changes to existing equipment" if is_change else "New registration"), + ], + link_path="/admin/reviews", + ) + + def register_equipment_pg( *, payload: dict[str, Any], @@ -1607,10 +1643,15 @@ def register_equipment_pg( ) cur.execute( - "SELECT id FROM equipment_registrations WHERE domain_id = %s", + # FOR UPDATE serializes concurrent submissions for this equipment; + # the previous status decides whether a new review cycle starts. + "SELECT id, status FROM equipment_registrations WHERE domain_id = %s FOR UPDATE", (domain_id,), ) existing_registration = cur.fetchone() + already_pending_review = bool( + existing_registration and existing_registration.get("status") == "pending" + ) features_count = len(features) if existing_registration: cur.execute( @@ -1683,7 +1724,24 @@ def register_equipment_pg( ) registration_row = cur.fetchone() + # Notify when a registration enters review. A resubmission while it is + # still pending coalesces into the open review the admins were + # already told about. + if record_status == "pending" and registration_row and not already_pending_review: + with conn.cursor() as cur: + _enqueue_equipment_review_notification( + cur, + registration_id=str(registration_row["id"]), + domain_id=domain_id, + equipment_name=str(payload.get("name") or domain_id), + user=user, + is_change=False, + ) conn.commit() + if record_status == "pending" and registration_row and not already_pending_review: + from platform_notifications import kick_notification_delivery + + kick_notification_delivery() return { "registration_id": str(registration_row["id"]) if registration_row else None, "domain_id": domain_id, @@ -1889,10 +1947,15 @@ def update_equipment_pg( ) cur.execute( - "SELECT id FROM equipment_registrations WHERE domain_id = %s", + # FOR UPDATE serializes concurrent submissions for this equipment; + # the previous status decides whether a new review cycle starts. + "SELECT id, status FROM equipment_registrations WHERE domain_id = %s FOR UPDATE", (domain_id,), ) existing_registration = cur.fetchone() + already_pending_review = bool( + existing_registration and existing_registration.get("status") == "pending" + ) if existing_registration: cur.execute( """ @@ -1952,7 +2015,24 @@ def update_equipment_pg( ) registration_row = cur.fetchone() + # Notify when a registration enters review. A resubmission while it is + # still pending coalesces into the open review the admins were + # already told about. + if record_status == "pending" and registration_row and not already_pending_review: + with conn.cursor() as cur: + _enqueue_equipment_review_notification( + cur, + registration_id=str(registration_row["id"]), + domain_id=domain_id, + equipment_name=str(payload.get("name") or domain_id), + user=user, + is_change=True, + ) conn.commit() + if record_status == "pending" and registration_row and not already_pending_review: + from platform_notifications import kick_notification_delivery + + kick_notification_delivery() return { "registration_id": str(registration_row["id"]) if registration_row else None, "domain_id": domain_id, diff --git a/api/platform_notifications.py b/api/platform_notifications.py new file mode 100644 index 0000000..ff6bd34 --- /dev/null +++ b/api/platform_notifications.py @@ -0,0 +1,587 @@ +""" +Platform notifications: a durable outbox delivered to Microsoft Teams and email. + +Events that need an administrator's attention (a user's first sign-in, an +equipment registration awaiting review, a role change request) call +``enqueue_notification`` with the cursor of the transaction that records the +event, so the notification exists exactly when the event does. A unique +``dedupe_key`` makes repeated calls for the same event harmless. + +Delivery runs outside request handling: ``kick_notification_delivery`` starts +a short background pass right after an event commits, and each API worker runs +a periodic sweeper. Each row is claimed on its own, just before it is sent, +with ``FOR UPDATE SKIP LOCKED``, a lease and a claim token; results are +recorded only while that claim is still held. Failures are retried with +backoff and marked failed after ``NOTIFICATION_MAX_ATTEMPTS``. Delivery is +at-least-once: a process that dies after sending but before recording the +result sends again. Error details stored or logged never include URLs or +credentials. + +Channels (each is skipped when not configured): + +- Teams: ``ADMIN_TEAMS_WEBHOOK_URL``, a Power Automate / Logic App flow that + posts the Adaptive Card it receives (the same pattern as the retrain alerts). +- Email, either + - SMTP: ``SMTP_HOST``, ``SMTP_PORT`` (587), ``SMTP_USERNAME``, + ``SMTP_PASSWORD``, ``SMTP_FROM``, ``SMTP_SECURITY`` (starttls | ssl | none); + - or ``EMAIL_WEBHOOK_URL``: a Power Automate flow that receives + ``{"to": [...], "subject", "text", "html"}`` and sends the message. + +Administrator recipients come from ``ADMIN_NOTIFICATION_EMAILS`` (comma +separated) or, when unset, every active admin in ``users``. ``PLATFORM_BASE_URL`` +turns ``link_path`` into a link in both channels. +""" +from __future__ import annotations + +import html +import json +import logging +import os +import smtplib +import ssl +import time +import urllib.error +import urllib.request +import uuid +from email.message import EmailMessage +from threading import Lock, Thread +from typing import Any, Iterable + +from psycopg2.extras import Json, RealDictCursor + +logger = logging.getLogger("dt.api.platform_notifications") + +AUDIENCE_ADMINS = "admins" +AUDIENCE_RECIPIENTS = "recipients" + +CHANNEL_PENDING = "pending" +CHANNEL_SENT = "sent" +CHANNEL_SKIPPED = "skipped" +CHANNEL_FAILED = "failed" + +_MAX_FACT_VALUE = 1000 +# One row is sent per claim; transports time out well within the lease. +_LEASE_SECONDS = 300 +SMTP_SECURITY_MODES = ("starttls", "ssl", "none") +_kick_lock = Lock() + + +def _env_int(name: str, default: int, minimum: int) -> int: + try: + return max(minimum, int(os.getenv(name, str(default)))) + except (TypeError, ValueError): + return default + + +def max_attempts() -> int: + return _env_int("NOTIFICATION_MAX_ATTEMPTS", 10, 1) + + +def _retry_delay_seconds(attempts: int) -> int: + # 1, 2, 4, ... minutes, capped at one hour. + return min(3600, 60 * (2 ** max(0, attempts - 1))) + + +# ── Schema ────────────────────────────────────────────────────────────────── + + +def ensure_platform_notifications_table_pg() -> None: + """Create the notification outbox (schema migration; owner connection).""" + from data_loader_pg import get_pg_migration_connection + + conn = get_pg_migration_connection() + try: + with conn.cursor() as cur: + cur.execute( + """ + CREATE TABLE IF NOT EXISTS platform_notifications ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + kind VARCHAR(80) NOT NULL, + dedupe_key VARCHAR(255) NOT NULL UNIQUE, + subject TEXT NOT NULL, + message TEXT NOT NULL, + facts_json JSONB NOT NULL DEFAULT '[]'::jsonb, + link_path TEXT NOT NULL DEFAULT '', + audience VARCHAR(20) NOT NULL DEFAULT 'admins', + recipient_emails JSONB NOT NULL DEFAULT '[]'::jsonb, + teams_status VARCHAR(20) NOT NULL DEFAULT 'pending', + email_status VARCHAR(20) NOT NULL DEFAULT 'pending', + attempts INTEGER NOT NULL DEFAULT 0, + next_attempt_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, + lease_expires_at TIMESTAMPTZ, + claim_token UUID, + email_retry_recipients JSONB NOT NULL DEFAULT '[]'::jsonb, + last_error TEXT NOT NULL DEFAULT '', + created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, + completed_at TIMESTAMPTZ + ) + """ + ) + conn.commit() + with conn.cursor() as cur: + # Columns added after the table first shipped. + cur.execute( + """ + ALTER TABLE platform_notifications + ADD COLUMN IF NOT EXISTS claim_token UUID, + ADD COLUMN IF NOT EXISTS email_retry_recipients JSONB NOT NULL DEFAULT '[]'::jsonb + """ + ) + cur.execute( + """ + CREATE INDEX IF NOT EXISTS idx_platform_notifications_open + ON platform_notifications (next_attempt_at) + WHERE completed_at IS NULL + """ + ) + # Request handlers enqueue inside their own transaction as + # api_client; only background delivery (api_worker) reads rows. + cur.execute("GRANT INSERT ON platform_notifications TO api_client") + conn.commit() + finally: + conn.close() + + +# ── Enqueue ───────────────────────────────────────────────────────────────── + + +def _clean_facts(facts: Iterable[tuple[str, Any]]) -> list[list[str]]: + cleaned: list[list[str]] = [] + for title, value in facts: + if value is None or value == "": + continue + text = value if isinstance(value, str) else json.dumps(value, default=str) + text = text.strip() + if len(text) > _MAX_FACT_VALUE: + text = text[: _MAX_FACT_VALUE - 1] + "…" + cleaned.append([str(title), text]) + return cleaned + + +def enqueue_notification( + cur, + *, + kind: str, + dedupe_key: str, + subject: str, + message: str, + facts: Iterable[tuple[str, Any]] = (), + link_path: str = "", + recipient_emails: Iterable[str] | None = None, +) -> bool: + """ + Record a notification in the caller's transaction. + + Without ``recipient_emails`` it goes to administrators over Teams and + email; with them, it is emailed to those addresses only. Returns False + when a notification with the same ``dedupe_key`` already exists. + + ``ON CONFLICT DO NOTHING`` has no conflict target on purpose: naming the + ``dedupe_key`` column would require SELECT on it, and api_client may only + insert into the outbox. The only other unique key is the random id. + """ + recipients = sorted({email.strip() for email in (recipient_emails or []) if email and email.strip()}) + # Email subjects must be a single line; names and equipment titles are user input. + subject = " ".join(str(subject).split()) + audience = AUDIENCE_RECIPIENTS if recipient_emails is not None else AUDIENCE_ADMINS + cur.execute( + """ + INSERT INTO platform_notifications ( + kind, dedupe_key, subject, message, facts_json, link_path, + audience, recipient_emails, teams_status + ) + VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s) + ON CONFLICT DO NOTHING + """, + ( + kind, + dedupe_key[:255], + subject[:300], + message, + Json(_clean_facts(facts)), + link_path, + audience, + Json(recipients), + CHANNEL_SKIPPED if audience == AUDIENCE_RECIPIENTS else CHANNEL_PENDING, + ), + ) + return bool(cur.rowcount) + + +# ── Rendering ─────────────────────────────────────────────────────────────── + + +def _platform_link(link_path: str) -> str: + base = os.getenv("PLATFORM_BASE_URL", "").strip().rstrip("/") + if not base or not link_path: + return "" + return f"{base}/{link_path.lstrip('/')}" + + +def build_teams_card(notification: dict[str, Any]) -> dict[str, Any]: + facts = notification.get("facts_json") or [] + card: dict[str, Any] = { + "$schema": "http://adaptivecards.io/schemas/adaptive-card.json", + "type": "AdaptiveCard", + "version": "1.4", + "msteams": {"width": "Full"}, + "body": [ + {"type": "TextBlock", "text": "Birck Digital Twin", "weight": "Bolder", "size": "Medium", "wrap": True}, + {"type": "TextBlock", "text": notification["subject"], "weight": "Bolder", "wrap": True}, + {"type": "TextBlock", "text": notification["message"], "wrap": True}, + {"type": "FactSet", "facts": [{"title": title, "value": value} for title, value in facts]}, + ], + } + link = _platform_link(notification.get("link_path") or "") + if link: + card["actions"] = [{"type": "Action.OpenUrl", "title": "Open in the platform", "url": link}] + return card + + +def build_email(notification: dict[str, Any]) -> tuple[str, str, str]: + """Return (subject, plain text, HTML); all user-supplied text is escaped in HTML.""" + subject = f"[Birck DT] {notification['subject']}" + facts = notification.get("facts_json") or [] + link = _platform_link(notification.get("link_path") or "") + text_lines = [notification["message"], ""] + text_lines += [f"{title}: {value}" for title, value in facts] + if link: + text_lines += ["", f"Open in the platform: {link}"] + rows = "".join( + f"{html.escape(title)}" + f"{html.escape(value)}" + for title, value in facts + ) + link_html = f'

Open in the platform

' if link else "" + body_html = ( + f"

{html.escape(notification['message'])}

" + f"{rows}
" + f"{link_html}" + ) + return subject, "\n".join(text_lines), body_html + + +# ── Channels ──────────────────────────────────────────────────────────────── + + +class ChannelNotConfigured(Exception): + """The channel has no configuration; the notification is skipped for it.""" + + +def _post_json(url: str, payload: dict[str, Any], *, timeout: int = 15) -> None: + request = urllib.request.Request( + url, + data=json.dumps(payload, default=str).encode("utf-8"), + headers={"Content-Type": "application/json"}, + method="POST", + ) + with urllib.request.urlopen(request, timeout=timeout) as response: + status = int(getattr(response, "status", 0) or response.getcode()) + if not 200 <= status < 300: + raise RuntimeError(f"webhook returned HTTP {status}") + + +def send_teams(notification: dict[str, Any]) -> None: + url = os.getenv("ADMIN_TEAMS_WEBHOOK_URL", "").strip() + if not url: + raise ChannelNotConfigured("ADMIN_TEAMS_WEBHOOK_URL is not set") + _post_json(url, build_teams_card(notification)) + + +def email_configured() -> bool: + return bool(os.getenv("SMTP_HOST", "").strip() or os.getenv("EMAIL_WEBHOOK_URL", "").strip()) + + +def send_email(recipients: list[str], notification: dict[str, Any]) -> list[str]: + """Send the email; return the recipients the server refused (to retry).""" + if not recipients: + raise ChannelNotConfigured("no email recipients") + subject, text, body_html = build_email(notification) + smtp_host = os.getenv("SMTP_HOST", "").strip() + if smtp_host: + sender = os.getenv("SMTP_FROM", "").strip() or os.getenv("SMTP_USERNAME", "").strip() + if not sender: + raise RuntimeError("SMTP_FROM is not set") + message = EmailMessage() + message["Subject"] = subject + message["From"] = sender + message["To"] = ", ".join(recipients) + message.set_content(text) + message.add_alternative(body_html, subtype="html") + port = _env_int("SMTP_PORT", 587, 1) + security = os.getenv("SMTP_SECURITY", "starttls").strip().lower() or "starttls" + if security not in SMTP_SECURITY_MODES: + # Never fall back to an unencrypted connection on a typo. + raise ValueError("SMTP_SECURITY must be starttls, ssl or none") + context = ssl.create_default_context() + if security == "ssl": + client: smtplib.SMTP = smtplib.SMTP_SSL(smtp_host, port, timeout=20, context=context) + else: + client = smtplib.SMTP(smtp_host, port, timeout=20) + with client: + if security == "starttls": + client.starttls(context=context) + username = os.getenv("SMTP_USERNAME", "").strip() + if username: + client.login(username, os.getenv("SMTP_PASSWORD", "")) + refused = client.send_message(message) or {} + return sorted(refused) + webhook = os.getenv("EMAIL_WEBHOOK_URL", "").strip() + if webhook: + _post_json(webhook, {"to": recipients, "subject": subject, "text": text, "html": body_html}) + return [] + raise ChannelNotConfigured("no email transport configured (SMTP_HOST or EMAIL_WEBHOOK_URL)") + + +def resolve_admin_emails(cur) -> list[str]: + configured = [ + email.strip() + for email in os.getenv("ADMIN_NOTIFICATION_EMAILS", "").split(",") + if email.strip() + ] + if configured: + return sorted(set(configured)) + cur.execute( + """ + SELECT DISTINCT email + FROM users + WHERE role = 'admin' + AND status = 'active' + AND email IS NOT NULL + AND email <> '' + AND email NOT LIKE '%@invalid.local' + ORDER BY email + """ + ) + return [row[0] if not isinstance(row, dict) else row["email"] for row in cur.fetchall()] + + +def describe_error(exc: BaseException) -> str: + """A short error description that never includes URLs, bodies or credentials.""" + if isinstance(exc, urllib.error.HTTPError): + return f"HTTP {exc.code}" + if isinstance(exc, urllib.error.URLError): + reason = exc.reason + return f"network error ({type(reason).__name__ if isinstance(reason, BaseException) else 'unreachable'})" + if isinstance(exc, smtplib.SMTPResponseException): + return f"{type(exc).__name__} {exc.smtp_code}" + if isinstance(exc, smtplib.SMTPRecipientsRefused): + return f"SMTPRecipientsRefused ({len(exc.recipients)} recipients)" + if isinstance(exc, (smtplib.SMTPException, OSError, ValueError, TimeoutError)): + return type(exc).__name__ + return type(exc).__name__ + + +# ── Delivery ──────────────────────────────────────────────────────────────── + + +def _claim(cur, limit: int = 1) -> list[dict[str, Any]]: + """Claim up to ``limit`` due rows under a fresh claim token (delivery uses 1).""" + cur.execute( + """ + WITH due AS ( + SELECT id + FROM platform_notifications + WHERE completed_at IS NULL + AND next_attempt_at <= CURRENT_TIMESTAMP + AND (lease_expires_at IS NULL OR lease_expires_at < CURRENT_TIMESTAMP) + ORDER BY created_at + FOR UPDATE SKIP LOCKED + LIMIT %s + ) + UPDATE platform_notifications n + SET attempts = n.attempts + 1, + lease_expires_at = CURRENT_TIMESTAMP + make_interval(secs => %s), + claim_token = %s::uuid + FROM due + WHERE n.id = due.id + RETURNING n.* + """, + (limit, _LEASE_SECONDS, str(uuid.uuid4())), + ) + return [dict(row) for row in cur.fetchall()] + + +def _deliver_one(notification: dict[str, Any], admin_emails: list[str]) -> dict[str, Any]: + statuses = { + "teams_status": notification["teams_status"], + "email_status": notification["email_status"], + } + errors: list[str] = [] + + if statuses["teams_status"] == CHANNEL_PENDING: + try: + send_teams(notification) + statuses["teams_status"] = CHANNEL_SENT + except ChannelNotConfigured: + statuses["teams_status"] = CHANNEL_SKIPPED + except Exception as exc: + errors.append(f"teams: {describe_error(exc)}") + + retry_recipients = list(notification.get("email_retry_recipients") or []) + if statuses["email_status"] == CHANNEL_PENDING: + # After a partial SMTP refusal, only the refused addresses are retried. + recipients = retry_recipients or ( + list(notification.get("recipient_emails") or []) + if notification.get("audience") == AUDIENCE_RECIPIENTS + else admin_emails + ) + try: + if not email_configured(): + raise ChannelNotConfigured("no email transport configured") + refused = send_email(recipients, notification) + if refused: + retry_recipients = refused + errors.append(f"email: {len(refused)} recipients refused") + else: + retry_recipients = [] + statuses["email_status"] = CHANNEL_SENT + except ChannelNotConfigured: + statuses["email_status"] = CHANNEL_SKIPPED + except Exception as exc: + errors.append(f"email: {describe_error(exc)}") + + pending = [name for name, value in statuses.items() if value == CHANNEL_PENDING] + exhausted = bool(pending) and int(notification["attempts"]) >= max_attempts() + if exhausted: + for name in pending: + statuses[name] = CHANNEL_FAILED + pending = [] + return { + **statuses, + "email_retry_recipients": retry_recipients, + "done": not pending, + "last_error": "; ".join(errors)[:2000], + "retry_in": _retry_delay_seconds(int(notification["attempts"])), + } + + +def _record_outcome(conn, notification: dict[str, Any], outcome: dict[str, Any]) -> bool: + """Store a delivery result; ignored if the claim was lost to another worker.""" + with conn.cursor() as cur: + cur.execute( + """ + UPDATE platform_notifications + SET teams_status = %s, + email_status = %s, + email_retry_recipients = %s, + last_error = %s, + lease_expires_at = NULL, + claim_token = NULL, + completed_at = CASE WHEN %s THEN CURRENT_TIMESTAMP ELSE NULL END, + next_attempt_at = CASE + WHEN %s THEN next_attempt_at + ELSE CURRENT_TIMESTAMP + make_interval(secs => %s) + END + WHERE id = %s AND claim_token = %s::uuid + """, + ( + outcome["teams_status"], + outcome["email_status"], + Json(outcome["email_retry_recipients"]), + outcome["last_error"], + outcome["done"], + outcome["done"], + outcome["retry_in"], + notification["id"], + str(notification["claim_token"]), + ), + ) + recorded = bool(cur.rowcount) + conn.commit() + return recorded + + +def _delete_old_notifications(conn) -> None: + days = _env_int("NOTIFICATION_RETENTION_DAYS", 90, 1) + with conn.cursor() as cur: + cur.execute( + """ + DELETE FROM platform_notifications + WHERE completed_at < CURRENT_TIMESTAMP - make_interval(days => %s) + """, + (days,), + ) + conn.commit() + + +def deliver_pending_notifications(*, limit: int = 20) -> dict[str, int]: + """Deliver up to ``limit`` due notifications, claiming each just before sending.""" + from data_loader_pg import get_pg_worker_connection + + conn = get_pg_worker_connection() + summary = {"claimed": 0, "completed": 0, "retrying": 0} + admin_emails: list[str] | None = None + try: + for _ in range(max(1, limit)): + with conn.cursor(cursor_factory=RealDictCursor) as cur: + claimed = _claim(cur, 1) + if claimed and admin_emails is None and claimed[0].get("audience") == AUDIENCE_ADMINS: + admin_emails = resolve_admin_emails(cur) + conn.commit() + if not claimed: + break + notification = claimed[0] + summary["claimed"] += 1 + outcome = _deliver_one(notification, admin_emails or []) + if not _record_outcome(conn, notification, outcome): + logger.warning("Notification %s claim was lost; result not recorded", notification["id"]) + continue + if outcome["done"]: + summary["completed"] += 1 + else: + summary["retrying"] += 1 + logger.warning( + "Notification %s (%s) will be retried: %s", + notification["id"], + notification["kind"], + outcome["last_error"], + ) + _delete_old_notifications(conn) + return summary + finally: + conn.close() + + +def kick_notification_delivery() -> None: + """Deliver new notifications soon; the sweeper covers anything this misses.""" + if os.getenv("USE_MOCK_DATABASE", "false").lower() == "true": + return + + # At most one fast-path pass per process; a pass already running (or the + # sweeper) picks up the new row. + if not _kick_lock.acquire(blocking=False): + return + + def _run() -> None: + try: + deliver_pending_notifications(limit=10) + except Exception: + logger.warning("Notification delivery pass failed; the sweeper will retry", exc_info=True) + finally: + _kick_lock.release() + + try: + Thread(target=_run, name="notification-delivery", daemon=True).start() + except Exception: + _kick_lock.release() + logger.warning("Could not start notification delivery thread", exc_info=True) + + +def start_notification_sweeper() -> Thread | None: + """Run delivery periodically in this process (NOTIFICATION_SWEEP_SECONDS, 0 disables).""" + interval = _env_int("NOTIFICATION_SWEEP_SECONDS", 60, 0) + if interval <= 0 or os.getenv("USE_MOCK_DATABASE", "false").lower() == "true": + return None + + def _loop() -> None: + while True: + try: + deliver_pending_notifications(limit=20) + except Exception: + logger.warning("Notification sweep failed", exc_info=True) + time.sleep(interval) + + thread = Thread(target=_loop, name="notification-sweeper", daemon=True) + thread.start() + return thread diff --git a/api/role_requests.py b/api/role_requests.py new file mode 100644 index 0000000..6e95369 --- /dev/null +++ b/api/role_requests.py @@ -0,0 +1,460 @@ +""" +Role change requests and the shared role-change rules. + +Everyone joins as a researcher. A user asks for a different role with a short +justification; administrators are notified (Teams, and email when configured) +and approve or reject it. Approval changes the role through +``apply_user_role_change``, the same rules and audit trail as the admin role +editor. The requester sees the decision on their Settings page (and is emailed +it when an email transport is configured), and the web app picks up the new +role on its next session read. +""" +from __future__ import annotations + +from typing import Any + +from psycopg2.extras import RealDictCursor + +from data_loader_pg import get_pg_connection, get_pg_migration_connection +from platform_notifications import enqueue_notification, kick_notification_delivery +from security import PlatformUser + +PLATFORM_ROLES = ("researcher", "pi", "equipment_owner", "admin") +ROLE_LABELS = { + "researcher": "Researcher", + "pi": "Principal Investigator", + "equipment_owner": "Equipment Owner", + "admin": "Administrator", +} +REQUEST_STATUSES = ("pending", "approved", "rejected", "cancelled") +MAX_JUSTIFICATION_LENGTH = 1000 +MAX_REVIEW_NOTE_LENGTH = 1000 +# Each request notifies every admin; cap how many one user can send per day. +MAX_REQUESTS_PER_DAY = 3 + + +class RoleChangeError(Exception): + """A role change or role request was refused; carries an HTTP status.""" + + def __init__(self, status_code: int, detail: str) -> None: + super().__init__(detail) + self.status_code = status_code + self.detail = detail + + +# ── Schema ────────────────────────────────────────────────────────────────── + + +def ensure_role_change_requests_table_pg() -> None: + """Create role_change_requests (schema migration; owner connection). + + Row-level security for the table lives in scripts/add_row_level_security.sql. + """ + conn = get_pg_migration_connection() + try: + with conn.cursor() as cur: + cur.execute( + """ + CREATE TABLE IF NOT EXISTS role_change_requests ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + user_id VARCHAR(255) NOT NULL REFERENCES users(id) ON DELETE CASCADE, + from_role VARCHAR(50) NOT NULL, + requested_role VARCHAR(50) NOT NULL, + justification TEXT NOT NULL DEFAULT '', + status VARCHAR(20) NOT NULL DEFAULT 'pending', + reviewer_id VARCHAR(255), + review_note TEXT NOT NULL DEFAULT '', + created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, + reviewed_at TIMESTAMPTZ, + updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, + CONSTRAINT role_change_requests_status_check + CHECK (status IN ('pending', 'approved', 'rejected', 'cancelled')), + CONSTRAINT role_change_requests_role_check + CHECK (requested_role IN ('researcher', 'pi', 'equipment_owner', 'admin')) + ) + """ + ) + # Enabled with the table, before any grant: until the policies in + # add_row_level_security.sql exist, api_client can see nothing. + cur.execute("ALTER TABLE role_change_requests ENABLE ROW LEVEL SECURITY") + conn.commit() + with conn.cursor() as cur: + # At most one open request per user. + cur.execute( + """ + CREATE UNIQUE INDEX IF NOT EXISTS uq_role_change_requests_pending + ON role_change_requests (user_id) + WHERE status = 'pending' + """ + ) + cur.execute( + """ + CREATE INDEX IF NOT EXISTS idx_role_change_requests_status + ON role_change_requests (status, created_at DESC) + """ + ) + cur.execute("GRANT SELECT, INSERT, UPDATE ON role_change_requests TO api_client") + conn.commit() + finally: + conn.close() + + +# ── Shared role-change rules ──────────────────────────────────────────────── + + +def apply_user_role_change( + cur, + *, + user_id: str, + new_role: str, + admin_user: PlatformUser, + reason: str, +) -> dict[str, Any]: + """ + Change a user's role in the caller's transaction and audit it. + + Refuses unknown roles, an admin demoting their own account, and demoting + the last administrator. Returns the (possibly unchanged) user row. + """ + new_role = (new_role or "").strip() + if new_role not in PLATFORM_ROLES: + raise RoleChangeError(400, "Invalid platform role") + cur.execute( + """ + SELECT id, name, email, role, organization, status, joined_at, last_active + FROM users + WHERE id = %s + FOR UPDATE + """, + (user_id,), + ) + existing = cur.fetchone() + if not existing: + raise RoleChangeError(404, "User not found") + + old_role = existing["role"] + if existing["id"] == admin_user.id and old_role == "admin" and new_role != "admin": + raise RoleChangeError(400, "Admins cannot demote their own account") + if old_role == "admin" and new_role != "admin": + cur.execute("SELECT COUNT(*) AS count FROM users WHERE role = 'admin'") + if int(cur.fetchone()["count"] or 0) <= 1: + raise RoleChangeError(400, "Cannot demote the last administrator") + + if old_role == new_role: + return dict(existing) + + cur.execute( + """ + UPDATE users + SET + role = %s, + -- Only a sign-in makes an account active; an invited user who + -- has not signed in yet stays invited (their first sign-in + -- notifies the admins). + status = CASE WHEN status IN ('disabled', 'invited') THEN status ELSE 'active' END + WHERE id = %s + RETURNING id, name, email, role, organization, status, joined_at, last_active + """, + (new_role, user_id), + ) + updated = dict(cur.fetchone()) + cur.execute( + """ + INSERT INTO user_role_audit ( + user_id, user_email, old_role, new_role, changed_by, changed_by_email, reason + ) + VALUES (%s, %s, %s, %s, %s, %s, %s) + """, + ( + existing["id"], + existing["email"] or "", + old_role, + new_role, + admin_user.id, + admin_user.email, + (reason or "").strip(), + ), + ) + return updated + + +# ── Requests ──────────────────────────────────────────────────────────────── + + +def _serialize(row: dict[str, Any]) -> dict[str, Any]: + record = dict(row) + record["id"] = str(record["id"]) + for key in ("created_at", "reviewed_at", "updated_at"): + if record.get(key) is not None: + record[key] = record[key].isoformat() + return record + + +_REQUEST_COLUMNS = """ + r.id, r.user_id, r.from_role, r.requested_role, r.justification, r.status, + r.reviewer_id, r.review_note, r.created_at, r.reviewed_at, r.updated_at, + u.name AS user_name, u.email AS user_email, u.organization AS user_organization +""" + + +def create_role_request_pg( + *, + user: PlatformUser, + requested_role: str, + justification: str, +) -> dict[str, Any]: + requested_role = (requested_role or "").strip() + justification = (justification or "").strip() + if requested_role not in PLATFORM_ROLES: + raise RoleChangeError(400, "Invalid platform role") + if not justification: + raise RoleChangeError(400, "Please explain why you need this role") + if len(justification) > MAX_JUSTIFICATION_LENGTH: + raise RoleChangeError(400, f"Justification must be at most {MAX_JUSTIFICATION_LENGTH} characters") + + conn = get_pg_connection(user) + try: + with conn.cursor(cursor_factory=RealDictCursor) as cur: + # Locking the user row serializes this user's concurrent requests, + # so the daily quota check and the insert below are atomic. + cur.execute( + "SELECT id, name, email, role, organization, status FROM users WHERE id = %s FOR UPDATE", + (user.id,), + ) + account = cur.fetchone() + if not account or account.get("status") in {"disabled", "merged"}: + raise RoleChangeError(403, "Your account cannot request a role change") + # The stored role is authoritative, not the session header. + if account["role"] == requested_role: + raise RoleChangeError(400, "You already have this role") + cur.execute( + """ + SELECT count(*) AS recent + FROM role_change_requests + WHERE user_id = %s + AND created_at > CURRENT_TIMESTAMP - interval '1 day' + """, + (user.id,), + ) + if int(cur.fetchone()["recent"] or 0) >= MAX_REQUESTS_PER_DAY: + raise RoleChangeError( + 429, + "You have sent several role requests today. Please try again tomorrow.", + ) + cur.execute( + """ + INSERT INTO role_change_requests (user_id, from_role, requested_role, justification) + VALUES (%s, %s, %s, %s) + ON CONFLICT (user_id) WHERE status = 'pending' DO NOTHING + RETURNING id + """, + (user.id, account["role"], requested_role, justification), + ) + inserted = cur.fetchone() + if not inserted: + raise RoleChangeError(409, "You already have a pending role request") + request_id = str(inserted["id"]) + enqueue_notification( + cur, + kind="role_change_requested", + dedupe_key=f"role_change_requested:{request_id}", + subject=f"Role request: {account['name'] or user.id} wants {ROLE_LABELS[requested_role]}", + message=( + f"{account['name'] or user.id} asked to change their role from " + f"{ROLE_LABELS.get(account['role'], account['role'])} to {ROLE_LABELS[requested_role]}." + ), + facts=[ + ("User", account["name"] or user.id), + ("Email", account["email"]), + ("Organization", account["organization"]), + ("Current role", ROLE_LABELS.get(account["role"], account["role"])), + ("Requested role", ROLE_LABELS[requested_role]), + ("Justification", justification), + ], + link_path="/admin/role-requests", + ) + cur.execute( + f""" + SELECT {_REQUEST_COLUMNS} + FROM role_change_requests r JOIN users u ON u.id = r.user_id + WHERE r.id = %s + """, + (request_id,), + ) + created = cur.fetchone() + conn.commit() + except Exception: + conn.rollback() + raise + finally: + conn.close() + kick_notification_delivery() + return _serialize(created) + + +def list_my_role_requests_pg(*, user: PlatformUser) -> list[dict[str, Any]]: + conn = get_pg_connection(user) + try: + with conn.cursor(cursor_factory=RealDictCursor) as cur: + cur.execute( + f""" + SELECT {_REQUEST_COLUMNS} + FROM role_change_requests r JOIN users u ON u.id = r.user_id + WHERE r.user_id = %s + ORDER BY r.created_at DESC + LIMIT 20 + """, + (user.id,), + ) + return [_serialize(row) for row in cur.fetchall()] + finally: + conn.close() + + +def cancel_role_request_pg(*, user: PlatformUser, request_id: str) -> dict[str, Any]: + conn = get_pg_connection(user) + try: + with conn.cursor(cursor_factory=RealDictCursor) as cur: + cur.execute( + """ + UPDATE role_change_requests + SET status = 'cancelled', updated_at = CURRENT_TIMESTAMP + WHERE id = %s AND user_id = %s AND status = 'pending' + RETURNING id + """, + (request_id, user.id), + ) + if not cur.fetchone(): + raise RoleChangeError(404, "No pending request to cancel") + cur.execute( + f""" + SELECT {_REQUEST_COLUMNS} + FROM role_change_requests r JOIN users u ON u.id = r.user_id + WHERE r.id = %s + """, + (request_id,), + ) + cancelled = cur.fetchone() + conn.commit() + return _serialize(cancelled) + except Exception: + conn.rollback() + raise + finally: + conn.close() + + +def list_role_requests_pg(*, admin_user: PlatformUser, status: str = "pending") -> list[dict[str, Any]]: + status = (status or "pending").strip() + if status != "all" and status not in REQUEST_STATUSES: + raise RoleChangeError(400, "Invalid request status filter") + conn = get_pg_connection(admin_user) + try: + with conn.cursor(cursor_factory=RealDictCursor) as cur: + where = "" if status == "all" else "WHERE r.status = %s" + cur.execute( + f""" + SELECT {_REQUEST_COLUMNS} + FROM role_change_requests r JOIN users u ON u.id = r.user_id + {where} + ORDER BY (r.status = 'pending') DESC, r.created_at DESC + LIMIT 200 + """, + () if status == "all" else (status,), + ) + return [_serialize(row) for row in cur.fetchall()] + finally: + conn.close() + + +def decide_role_request_pg( + *, + admin_user: PlatformUser, + request_id: str, + decision: str, + note: str = "", +) -> dict[str, Any]: + decision = (decision or "").strip().lower() + if decision not in {"approve", "reject"}: + raise RoleChangeError(400, "Decision must be approve or reject") + note = (note or "").strip() + if len(note) > MAX_REVIEW_NOTE_LENGTH: + raise RoleChangeError(400, f"Note must be at most {MAX_REVIEW_NOTE_LENGTH} characters") + + conn = get_pg_connection(admin_user) + try: + with conn.cursor(cursor_factory=RealDictCursor) as cur: + cur.execute( + f""" + SELECT {_REQUEST_COLUMNS} + FROM role_change_requests r JOIN users u ON u.id = r.user_id + WHERE r.id = %s + FOR UPDATE OF r + """, + (request_id,), + ) + request = cur.fetchone() + if not request: + raise RoleChangeError(404, "Role request not found") + if request["status"] != "pending": + raise RoleChangeError(409, f"This request is already {request['status']}") + + new_status = "approved" if decision == "approve" else "rejected" + if decision == "approve": + apply_user_role_change( + cur, + user_id=request["user_id"], + new_role=request["requested_role"], + admin_user=admin_user, + reason=f"Role request {request_id}" + (f": {note}" if note else ""), + ) + cur.execute( + """ + UPDATE role_change_requests + SET status = %s, + reviewer_id = %s, + review_note = %s, + reviewed_at = CURRENT_TIMESTAMP, + updated_at = CURRENT_TIMESTAMP + WHERE id = %s + """, + (new_status, admin_user.id, note, request_id), + ) + if request.get("user_email"): + label = ROLE_LABELS.get(request["requested_role"], request["requested_role"]) + enqueue_notification( + cur, + kind="role_change_decided", + dedupe_key=f"role_change_decided:{request_id}", + subject=( + f"Your request for the {label} role was approved" + if new_status == "approved" + else f"Your request for the {label} role was not approved" + ), + message=( + f"An administrator approved your request. You now have the {label} role; " + "reload the platform to see the change." + if new_status == "approved" + else "An administrator reviewed your request and did not approve it." + ), + facts=[("Requested role", label), ("Note from the administrator", note)], + link_path="/settings", + recipient_emails=[request["user_email"]], + ) + cur.execute( + f""" + SELECT {_REQUEST_COLUMNS} + FROM role_change_requests r JOIN users u ON u.id = r.user_id + WHERE r.id = %s + """, + (request_id,), + ) + decided = cur.fetchone() + conn.commit() + except Exception: + conn.rollback() + raise + finally: + conn.close() + kick_notification_delivery() + return _serialize(decided) diff --git a/api/routers/admin.py b/api/routers/admin.py index 8813b3d..ea4a12a 100644 --- a/api/routers/admin.py +++ b/api/routers/admin.py @@ -1,7 +1,7 @@ import logging from typing import List, Dict, Any -from fastapi import APIRouter, Depends, HTTPException +from fastapi import APIRouter, Depends, Header, HTTPException from psycopg2.extras import RealDictCursor from pydantic import BaseModel, EmailStr from data_loader_pg import get_pg_connection, get_pg_worker_connection @@ -20,12 +20,24 @@ require_ingestion_token, require_system_token, ) +from platform_notifications import enqueue_notification, kick_notification_delivery +from role_requests import ( + PLATFORM_ROLES as ROLE_REQUEST_ROLES, + ROLE_LABELS, + RoleChangeError, + apply_user_role_change, + cancel_role_request_pg, + create_role_request_pg, + decide_role_request_pg, + list_my_role_requests_pg, + list_role_requests_pg, +) from user_identity import sync_canonical_user logger = logging.getLogger(__name__) router = APIRouter(prefix="/admin", tags=["admin"]) -PLATFORM_ROLES = {"admin", "pi", "equipment_owner", "researcher"} +PLATFORM_ROLES = set(ROLE_REQUEST_ROLES) @router.get("/users/{email}/role") def get_or_create_user_role( @@ -33,12 +45,20 @@ def get_or_create_user_role( name: str | None = None, organization: str | None = None, user_id: str | None = None, + x_user_id: str | None = Header(None, alias="X-User-Id"), _: None = Depends(require_system_token), ): """ Called by Next-Auth during sign-in. Returns the user's role. If the user does not exist, registers them automatically as 'researcher'. + + Server-to-server only: the sign-in code sends just the system token, + while the browser proxy and MCP bridge always add X-User-* headers. A call + carrying a user identity would let a signed-in user create or merge + someone else's account (and fake their first-sign-in notification). """ + if x_user_id is not None: + raise HTTPException(status_code=403, detail="Not available through the user proxy") try: conn = get_pg_worker_connection() with conn.cursor(cursor_factory=RealDictCursor) as cur: @@ -50,7 +70,31 @@ def get_or_create_user_role( organization=organization, requested_status="active", ) + notified = False + if user.get("first_sign_in"): + # This endpoint runs on every session read; the dedupe key + # makes the notification fire once per user. + notified = enqueue_notification( + cur, + kind="user_first_sign_in", + dedupe_key=f"user_first_sign_in:{user['id']}", + subject=f"New user signed in: {user['name'] or user['id']}", + message=( + f"{user['name'] or user['id']} signed in to the platform for the first time " + f"and has the {ROLE_LABELS.get(user['role'], user['role'])} role." + ), + facts=[ + ("User", user["name"] or user["id"]), + ("Email", user["email"]), + ("Organization", user["organization"]), + ("Role", ROLE_LABELS.get(user["role"], user["role"])), + ("User ID", user["id"]), + ], + link_path="/admin/users", + ) conn.commit() + if notified: + kick_notification_delivery() response = { "role": user["role"], "organization": user["organization"] or organization or "Purdue University", @@ -162,80 +206,22 @@ def update_user_role( admin_user: PlatformUser = Depends(require_admin), ) -> Dict[str, Any]: """Promote or demote an existing platform user after they self-register.""" - new_role = payload.role.strip() - if new_role not in PLATFORM_ROLES: - raise HTTPException(status_code=400, detail="Invalid platform role") - try: - conn = get_pg_connection() + conn = get_pg_connection(admin_user) with conn.cursor(cursor_factory=RealDictCursor) as cur: - cur.execute( - """ - SELECT id, name, email, role, organization, status, joined_at, last_active - FROM users - WHERE id = %s - FOR UPDATE - """, - (user_id,), + updated = apply_user_role_change( + cur, + user_id=user_id, + new_role=payload.role, + admin_user=admin_user, + reason=payload.reason, ) - existing = cur.fetchone() - if not existing: - raise HTTPException(status_code=404, detail="User not found") - - old_role = existing["role"] - if existing["id"] == admin_user.id and old_role == "admin" and new_role != "admin": - raise HTTPException(status_code=400, detail="Admins cannot demote their own account") - - if old_role == "admin" and new_role != "admin": - cur.execute("SELECT COUNT(*) AS count FROM users WHERE role = 'admin'") - admin_count = int(cur.fetchone()["count"] or 0) - if admin_count <= 1: - raise HTTPException(status_code=400, detail="Cannot demote the last administrator") - - if old_role != new_role: - cur.execute( - """ - UPDATE users - SET - role = %s, - status = CASE WHEN status = 'disabled' THEN status ELSE 'active' END - WHERE id = %s - RETURNING id, name, email, role, organization, status, joined_at, last_active - """, - (new_role, user_id), - ) - updated = cur.fetchone() - cur.execute( - """ - INSERT INTO user_role_audit ( - user_id, - user_email, - old_role, - new_role, - changed_by, - changed_by_email, - reason - ) - VALUES (%s, %s, %s, %s, %s, %s, %s) - """, - ( - existing["id"], - existing["email"] or "", - old_role, - new_role, - admin_user.id, - admin_user.email, - payload.reason.strip(), - ), - ) - else: - updated = existing conn.commit() return _serialize_user_record(updated) - except HTTPException: + except RoleChangeError as exc: if 'conn' in locals() and conn: conn.rollback() - raise + raise HTTPException(status_code=exc.status_code, detail=exc.detail) from exc except Exception as e: if 'conn' in locals() and conn: conn.rollback() @@ -246,6 +232,78 @@ def update_user_role( conn.close() +# ── Role change requests ───────────────────────────────────────────────────── + + +class RoleRequestPayload(BaseModel): + requested_role: str + justification: str = "" + + +class RoleRequestDecisionPayload(BaseModel): + decision: str + note: str = "" + + +def _role_request_call(fn, **kwargs): + try: + return fn(**kwargs) + except RoleChangeError as exc: + raise HTTPException(status_code=exc.status_code, detail=exc.detail) from exc + + +@router.post("/role-requests") +def create_role_request( + payload: RoleRequestPayload, + user: PlatformUser = Depends(get_platform_user), +) -> Dict[str, Any]: + """Ask administrators for a different platform role.""" + return _role_request_call( + create_role_request_pg, + user=user, + requested_role=payload.requested_role, + justification=payload.justification, + ) + + +@router.get("/role-requests/mine") +def my_role_requests( + user: PlatformUser = Depends(get_platform_user), +) -> List[Dict[str, Any]]: + return _role_request_call(list_my_role_requests_pg, user=user) + + +@router.post("/role-requests/{request_id}/cancel") +def cancel_role_request( + request_id: str, + user: PlatformUser = Depends(get_platform_user), +) -> Dict[str, Any]: + return _role_request_call(cancel_role_request_pg, user=user, request_id=request_id) + + +@router.get("/role-requests") +def list_role_requests( + status: str = "pending", + admin_user: PlatformUser = Depends(require_admin), +) -> List[Dict[str, Any]]: + return _role_request_call(list_role_requests_pg, admin_user=admin_user, status=status) + + +@router.post("/role-requests/{request_id}/decision") +def decide_role_request( + request_id: str, + payload: RoleRequestDecisionPayload, + admin_user: PlatformUser = Depends(require_admin), +) -> Dict[str, Any]: + return _role_request_call( + decide_role_request_pg, + admin_user=admin_user, + request_id=request_id, + decision=payload.decision, + note=payload.note, + ) + + @router.get("/reviews") def get_reviews( _: PlatformUser = Depends(require_admin), diff --git a/api/scripts/add_row_level_security.sql b/api/scripts/add_row_level_security.sql index 01036e8..ebcbc8c 100644 --- a/api/scripts/add_row_level_security.sql +++ b/api/scripts/add_row_level_security.sql @@ -746,3 +746,30 @@ CREATE POLICY dt_recipe_proposals_update ON experiment_recipe_proposals DROP POLICY IF EXISTS dt_recipe_proposals_worker ON experiment_recipe_proposals; CREATE POLICY dt_recipe_proposals_worker ON experiment_recipe_proposals FOR ALL TO api_worker USING (true) WITH CHECK (true); + +-- ── Policies: role change requests ───────────────────────────────────────── +-- Always enforced, like project_members. Users create and cancel their own +-- requests; only administrators see everyone's and decide them. + +DROP POLICY IF EXISTS dt_role_requests_select ON role_change_requests; +CREATE POLICY dt_role_requests_select ON role_change_requests + FOR SELECT USING (app_is_admin() OR user_id = app_current_user_id()); +DROP POLICY IF EXISTS dt_role_requests_insert ON role_change_requests; +CREATE POLICY dt_role_requests_insert ON role_change_requests + FOR INSERT WITH CHECK ( + user_id = app_current_user_id() + AND status = 'pending' + AND reviewer_id IS NULL + ); +DROP POLICY IF EXISTS dt_role_requests_admin_update ON role_change_requests; +CREATE POLICY dt_role_requests_admin_update ON role_change_requests + FOR UPDATE USING (app_is_admin()) WITH CHECK (app_is_admin()); +DROP POLICY IF EXISTS dt_role_requests_cancel_own ON role_change_requests; +CREATE POLICY dt_role_requests_cancel_own ON role_change_requests + FOR UPDATE + USING (user_id = app_current_user_id() AND status = 'pending') + WITH CHECK (user_id = app_current_user_id() AND status = 'cancelled' AND reviewer_id IS NULL); +DROP POLICY IF EXISTS dt_role_requests_worker ON role_change_requests; +CREATE POLICY dt_role_requests_worker ON role_change_requests + FOR ALL TO api_worker USING (true) WITH CHECK (true); +ALTER TABLE role_change_requests ENABLE ROW LEVEL SECURITY; diff --git a/api/tests/test_admin_notifications_pg.py b/api/tests/test_admin_notifications_pg.py new file mode 100644 index 0000000..cf9f447 --- /dev/null +++ b/api/tests/test_admin_notifications_pg.py @@ -0,0 +1,705 @@ +""" +Admin notifications and role change requests against a disposable Postgres. + +Same harness and safety rules as test_row_level_security_pg.py: skipped unless +DT_RLS_TEST_PG_PORT is set, and refuses to run against a database without the +``rls_test_marker`` table. +""" +from __future__ import annotations + +import json +import os +import sys +import unittest +import uuid +from pathlib import Path +from unittest.mock import patch + +API_ROOT = Path(__file__).resolve().parents[1] +if str(API_ROOT) not in sys.path: + sys.path.insert(0, str(API_ROOT)) + +TEST_PORT = os.getenv("DT_RLS_TEST_PG_PORT", "").strip() + +if TEST_PORT: + os.environ["PG_HOST"] = os.getenv("DT_RLS_TEST_PG_HOST", "127.0.0.1") + os.environ["PG_PORT"] = TEST_PORT + os.environ["PG_DB"] = os.getenv("DT_RLS_TEST_PG_DB", "digital_twin") + os.environ["PG_USER"] = "api_client" + os.environ["PG_PASS"] = os.environ["DT_RLS_TEST_API_PASS"] + os.environ["PG_SUPER_USER"] = os.getenv("DT_RLS_TEST_SUPER_USER", "postgres") + os.environ["PG_SUPER_PASS"] = os.environ["DT_RLS_TEST_SUPER_PASS"] + os.environ["PG_WORKER_USER"] = "api_worker" + os.environ["PG_WORKER_PASS"] = os.environ["DT_RLS_TEST_WORKER_PASS"] + os.environ["USE_MOCK_DATABASE"] = "false" + os.environ.setdefault("DT_SYSTEM_TOKEN", "rls-test-system-token") + os.environ.setdefault("INGESTION_TOKEN", "rls-test-ingestion-token") + +import psycopg2 # noqa: E402 +import psycopg2.errors # noqa: E402 +import psycopg2.extras # noqa: E402 + +from security import PlatformUser # noqa: E402 + +NOTIFY_ENV_KEYS = ( + "ADMIN_TEAMS_WEBHOOK_URL", + "EMAIL_WEBHOOK_URL", + "SMTP_HOST", + "ADMIN_NOTIFICATION_EMAILS", + "PLATFORM_BASE_URL", +) + + +def _user(user_id: str, role: str, org: str = "Purdue University") -> PlatformUser: + return PlatformUser(id=user_id, email=f"{user_id}@example.test", name=user_id.split("-")[0].title(), role=role, organization=org) + + +class _FakeResponse: + status = 200 + + def __enter__(self): + return self + + def __exit__(self, *exc): + return False + + def getcode(self): + return 200 + + +@unittest.skipUnless(TEST_PORT, "set DT_RLS_TEST_PG_PORT to run database integration tests") +class AdminNotificationIntegrationTests(unittest.TestCase): + @classmethod + def setUpClass(cls): + import data_loader_pg + import db_migrations + + conn = data_loader_pg.get_pg_migration_connection() + try: + with conn.cursor() as cur: + cur.execute("SELECT to_regclass('public.rls_test_marker')") + if cur.fetchone()[0] is None: + raise RuntimeError("Refusing to run: the target database has no rls_test_marker table") + finally: + conn.close() + db_migrations.run_schema_migrations() + + suffix = uuid.uuid4().hex[:6] + cls.suffix = suffix + cls.admin = _user(f"admin-{suffix}", "admin") + cls.researcher = _user(f"res-{suffix}", "researcher") + cls.other = _user(f"oth-{suffix}", "researcher") + cls.owner = _user(f"own-{suffix}", "equipment_owner") + conn = data_loader_pg.get_pg_migration_connection() + try: + with conn.cursor() as cur: + for person in (cls.admin, cls.researcher, cls.other, cls.owner): + cur.execute( + "INSERT INTO users (id, name, email, role, organization, status) VALUES (%s, %s, %s, %s, %s, 'active')", + (person.id, person.name, person.email, person.role, person.organization), + ) + conn.commit() + finally: + conn.close() + + def setUp(self): + self._env = patch.dict(os.environ, {key: "" for key in NOTIFY_ENV_KEYS}) + self._env.start() + self.addCleanup(self._env.stop) + # Delivery is exercised explicitly; keep request paths from starting threads. + kick = patch("platform_notifications.kick_notification_delivery") + self.addCleanup(kick.stop) + kick.start() + for module in ("role_requests", "routers.admin"): + p = patch(f"{module}.kick_notification_delivery") + self.addCleanup(p.stop) + p.start() + + # ── helpers ────────────────────────────────────────────────────────── + + def _owner_sql(self, sql: str, params: tuple = ()): + import data_loader_pg + + conn = data_loader_pg.get_pg_migration_connection() + try: + with conn.cursor() as cur: + cur.execute(sql, params) + rows = cur.fetchall() if cur.description else [] + conn.commit() + return rows + finally: + conn.close() + + def _notifications(self, kind: str, like: str) -> list[tuple]: + return self._owner_sql( + "SELECT id, dedupe_key, audience, recipient_emails, teams_status, email_status, facts_json " + "FROM platform_notifications WHERE kind = %s AND dedupe_key LIKE %s ORDER BY created_at", + (kind, like), + ) + + def _client(self): + from fastapi import FastAPI + from fastapi.testclient import TestClient + from routers import admin, equipment + + app = FastAPI() + app.include_router(admin.router, prefix="/api") + app.include_router(equipment.router, prefix="/api") + return TestClient(app) + + def _headers(self, person: PlatformUser) -> dict[str, str]: + return { + "X-System-Token": os.environ["DT_SYSTEM_TOKEN"], + "X-User-Id": person.id, + "X-User-Email": person.email, + "X-User-Name": person.name, + "X-User-Role": person.role, + "X-User-Org": person.organization, + } + + # ── first sign-in ─────────────────────────────────────────────────── + + def test_first_sign_in_notifies_once(self): + client = self._client() + email = f"newbie-{self.suffix}@example.test" + headers = {"X-System-Token": os.environ["DT_SYSTEM_TOKEN"]} + for _ in range(3): # the web app calls this on every session read + response = client.get(f"/api/admin/users/{email}/role", headers=headers, params={"name": "Newbie"}) + self.assertEqual(response.status_code, 200, response.text) + rows = self._notifications("user_first_sign_in", f"user_first_sign_in:newbie-{self.suffix}%") + self.assertEqual(len(rows), 1) + facts = dict(rows[0][6]) + self.assertEqual(facts["Email"], email) + self.assertEqual(facts["Role"], "Researcher") + + def test_invited_user_first_sign_in_notifies_but_existing_user_does_not(self): + client = self._client() + invited_email = f"invitee-{self.suffix}@example.com" + response = client.post( + "/api/admin/users", + headers=self._headers(self.admin), + json={"email": invited_email, "role": "pi", "name": "Invitee"}, + ) + self.assertEqual(response.status_code, 200, response.text) + self.assertEqual(self._notifications("user_first_sign_in", f"user_first_sign_in:invitee-{self.suffix}%"), []) + client.get(f"/api/admin/users/{invited_email}/role", headers={"X-System-Token": os.environ["DT_SYSTEM_TOKEN"]}) + self.assertEqual(len(self._notifications("user_first_sign_in", f"user_first_sign_in:invitee-{self.suffix}%")), 1) + + client.get(f"/api/admin/users/{self.researcher.email}/role", headers={"X-System-Token": os.environ["DT_SYSTEM_TOKEN"]}) + self.assertEqual(self._notifications("user_first_sign_in", f"user_first_sign_in:{self.researcher.id}"), []) + + # ── equipment awaiting review ─────────────────────────────────────── + + def test_equipment_registration_awaiting_review_notifies_admins(self): + from metadata_pg import register_equipment_pg + + name = f"Review Etcher {self.suffix}" + result = register_equipment_pg( + payload={"name": name, "columns": [{"name": "pressure", "type": "float"}], "primary_target": "rate"}, + user=self.owner, + ) + self.assertEqual(result["status"], "pending") + rows = self._notifications("equipment_review_requested", f"equipment_review_requested:{result['registration_id']}:%") + self.assertEqual(len(rows), 1) + self.assertEqual(dict(rows[0][6])["Equipment"], name) + + admin_result = register_equipment_pg( + payload={"name": f"Admin Etcher {self.suffix}", "columns": [{"name": "p", "type": "float"}], "primary_target": "rate"}, + user=self.admin, + ) + self.assertEqual(admin_result["status"], "approved") + self.assertEqual( + self._notifications("equipment_review_requested", f"equipment_review_requested:{admin_result['registration_id']}:%"), + [], + ) + + # ── role change requests ──────────────────────────────────────────── + + def test_role_request_lifecycle_with_notifications(self): + client = self._client() + requester = _user(f"req-{uuid.uuid4().hex[:6]}", "researcher") + self._owner_sql( + "INSERT INTO users (id, name, email, role, organization, status) VALUES (%s, %s, %s, 'researcher', %s, 'active')", + (requester.id, requester.name, requester.email, requester.organization), + ) + + same = client.post("/api/admin/role-requests", headers=self._headers(requester), json={"requested_role": "researcher", "justification": "x"}) + self.assertEqual(same.status_code, 400) + empty = client.post("/api/admin/role-requests", headers=self._headers(requester), json={"requested_role": "pi", "justification": " "}) + self.assertEqual(empty.status_code, 400) + + created = client.post( + "/api/admin/role-requests", + headers=self._headers(requester), + json={"requested_role": "pi", "justification": "I lead the etch project."}, + ) + self.assertEqual(created.status_code, 200, created.text) + request_id = created.json()["id"] + self.assertEqual(created.json()["from_role"], "researcher") + duplicate = client.post( + "/api/admin/role-requests", + headers=self._headers(requester), + json={"requested_role": "equipment_owner", "justification": "again"}, + ) + self.assertEqual(duplicate.status_code, 409) + rows = self._notifications("role_change_requested", f"role_change_requested:{request_id}") + self.assertEqual(len(rows), 1) + self.assertEqual(dict(rows[0][6])["Justification"], "I lead the etch project.") + + # Only admins list everyone's requests. + self.assertEqual(client.get("/api/admin/role-requests", headers=self._headers(requester)).status_code, 403) + pending = client.get("/api/admin/role-requests", headers=self._headers(self.admin)).json() + self.assertIn(request_id, {row["id"] for row in pending}) + mine = client.get("/api/admin/role-requests/mine", headers=self._headers(requester)).json() + self.assertEqual([row["id"] for row in mine], [request_id]) + + approved = client.post( + f"/api/admin/role-requests/{request_id}/decision", + headers=self._headers(self.admin), + json={"decision": "approve", "note": "Welcome aboard"}, + ) + self.assertEqual(approved.status_code, 200, approved.text) + self.assertEqual(approved.json()["status"], "approved") + self.assertEqual(self._owner_sql("SELECT role FROM users WHERE id = %s", (requester.id,))[0][0], "pi") + audit = self._owner_sql( + "SELECT old_role, new_role, changed_by, reason FROM user_role_audit WHERE user_id = %s", (requester.id,) + ) + self.assertEqual(audit[0][:3], ("researcher", "pi", self.admin.id)) + self.assertIn(request_id, audit[0][3]) + decided = self._notifications("role_change_decided", f"role_change_decided:{request_id}") + self.assertEqual(len(decided), 1) + self.assertEqual((decided[0][2], decided[0][3], decided[0][4]), ("recipients", [requester.email], "skipped")) + + again = client.post( + f"/api/admin/role-requests/{request_id}/decision", + headers=self._headers(self.admin), + json={"decision": "reject"}, + ) + self.assertEqual(again.status_code, 409) + + def test_cancel_then_request_again_and_reject(self): + from role_requests import cancel_role_request_pg, create_role_request_pg, decide_role_request_pg + + first = create_role_request_pg(user=self.other, requested_role="equipment_owner", justification="facility work") + cancel_role_request_pg(user=self.other, request_id=first["id"]) + second = create_role_request_pg(user=self.other, requested_role="pi", justification="new project") + rejected = decide_role_request_pg(admin_user=self.admin, request_id=second["id"], decision="reject", note="Not yet") + self.assertEqual(rejected["status"], "rejected") + self.assertEqual(self._owner_sql("SELECT role FROM users WHERE id = %s", (self.other.id,))[0][0], "researcher") + + def test_role_request_rls(self): + from data_loader_pg import get_pg_connection + from role_requests import create_role_request_pg + + mine = create_role_request_pg(user=self.researcher, requested_role="pi", justification="my request") + conn = get_pg_connection(self.other) + try: + with conn.cursor() as cur: + cur.execute("SELECT count(*) FROM role_change_requests WHERE id = %s", (mine["id"],)) + self.assertEqual(cur.fetchone()[0], 0) + cur.execute( + "UPDATE role_change_requests SET status = 'cancelled' WHERE id = %s", (mine["id"],) + ) + self.assertEqual(cur.rowcount, 0) + with self.assertRaises(psycopg2.errors.InsufficientPrivilege): + cur.execute( + "INSERT INTO role_change_requests (user_id, from_role, requested_role) VALUES (%s, 'researcher', 'admin')", + (self.researcher.id,), + ) + conn.rollback() + finally: + conn.close() + # The requester cannot approve their own request. + conn = get_pg_connection(self.researcher) + try: + with conn.cursor() as cur: + with self.assertRaises(psycopg2.errors.InsufficientPrivilege): + cur.execute( + "UPDATE role_change_requests SET status = 'approved' WHERE id = %s", (mine["id"],) + ) + conn.rollback() + finally: + conn.close() + # api_client can enqueue but never read the outbox. + conn = get_pg_connection(self.admin) + try: + with conn.cursor() as cur: + with self.assertRaises(psycopg2.errors.InsufficientPrivilege): + cur.execute("SELECT count(*) FROM platform_notifications") + finally: + conn.close() + + # ── delivery ──────────────────────────────────────────────────────── + + def _fresh_notification(self, **kwargs) -> str: + import data_loader_pg + from platform_notifications import enqueue_notification + + key = f"test:{uuid.uuid4().hex}" + conn = data_loader_pg.get_pg_worker_connection() + try: + with conn.cursor() as cur: + enqueue_notification( + cur, + kind="test", + dedupe_key=key, + subject="Test ", + message="Something happened", + facts=[("Who", "O'Brien ")], + link_path="/admin/users", + **kwargs, + ) + conn.commit() + finally: + conn.close() + # Only this test's row is due. + self._owner_sql( + "UPDATE platform_notifications SET next_attempt_at = now() + interval '1 day' " + "WHERE completed_at IS NULL AND dedupe_key <> %s", + (key,), + ) + return key + + def _state(self, key: str) -> tuple: + return self._owner_sql( + "SELECT teams_status, email_status, attempts, completed_at IS NOT NULL, last_error, next_attempt_at > now() " + "FROM platform_notifications WHERE dedupe_key = %s", + (key,), + )[0] + + def test_delivery_to_teams_and_email_webhook(self): + from platform_notifications import deliver_pending_notifications + + key = self._fresh_notification() + posted = [] + + def fake_urlopen(request, timeout=0): + posted.append((request.full_url, json.loads(request.data))) + return _FakeResponse() + + env = { + "ADMIN_TEAMS_WEBHOOK_URL": "https://teams.example.test/hook", + "EMAIL_WEBHOOK_URL": "https://mail.example.test/hook", + "PLATFORM_BASE_URL": "https://dt.example.test", + } + with patch.dict(os.environ, env), patch("platform_notifications.urllib.request.urlopen", side_effect=fake_urlopen): + summary = deliver_pending_notifications() + self.assertEqual(summary["completed"], 1) + self.assertEqual(self._state(key)[:4], ("sent", "sent", 1, True)) + + teams = next(body for url, body in posted if url.startswith("https://teams")) + self.assertEqual(teams["type"], "AdaptiveCard") + self.assertEqual(teams["actions"][0]["url"], "https://dt.example.test/admin/users") + email = next(body for url, body in posted if url.startswith("https://mail")) + self.assertIn(self.admin.email, email["to"]) + self.assertNotIn(self.researcher.email, email["to"]) + self.assertIn("O'Brien <x>", email["html"]) + self.assertIn("Something <b>happened</b>", email["html"]) + + def test_explicit_admin_list_and_smtp(self): + from platform_notifications import deliver_pending_notifications + + key = self._fresh_notification() + sent = [] + + class FakeSMTP: + def __init__(self, host, port, timeout=0): + self.host, self.port = host, port + + def __enter__(self): + return self + + def __exit__(self, *exc): + return False + + def starttls(self, context=None): + sent.append("starttls") + + def login(self, user, password): + sent.append(("login", user)) + + def send_message(self, message): + sent.append(message) + + env = {"SMTP_HOST": "smtp.example.test", "SMTP_FROM": "dt@example.test", "SMTP_USERNAME": "dt", + "SMTP_PASSWORD": "pw", "ADMIN_NOTIFICATION_EMAILS": "a@example.test, b@example.test"} + with patch.dict(os.environ, env), patch("platform_notifications.smtplib.SMTP", FakeSMTP): + deliver_pending_notifications() + self.assertEqual(self._state(key)[:2], ("skipped", "sent")) + message = [item for item in sent if not isinstance(item, (str, tuple))][0] + self.assertEqual(message["To"], "a@example.test, b@example.test") + self.assertTrue(message["Subject"].startswith("[Birck DT] ")) + self.assertIn("starttls", sent) + + def test_unconfigured_channels_are_skipped(self): + from platform_notifications import deliver_pending_notifications + + key = self._fresh_notification() + deliver_pending_notifications() + self.assertEqual(self._state(key)[:4], ("skipped", "skipped", 1, True)) + + def test_failures_retry_with_backoff_then_give_up(self): + from platform_notifications import deliver_pending_notifications + + key = self._fresh_notification() + env = {"ADMIN_TEAMS_WEBHOOK_URL": "https://teams.example.test/hook", "NOTIFICATION_MAX_ATTEMPTS": "2"} + with patch.dict(os.environ, env), patch( + "platform_notifications.urllib.request.urlopen", side_effect=OSError("network down") + ): + deliver_pending_notifications() + teams, email, attempts, done, error, backed_off = self._state(key) + self.assertEqual((teams, email, attempts, done, backed_off), ("pending", "skipped", 1, False, True)) + self.assertEqual(error, "teams: OSError") + # Not due yet: nothing is claimed. + self.assertEqual(deliver_pending_notifications()["claimed"], 0) + self._owner_sql("UPDATE platform_notifications SET next_attempt_at = now() WHERE dedupe_key = %s", (key,)) + deliver_pending_notifications() + self.assertEqual(self._state(key)[:4], ("failed", "skipped", 2, True)) + + def test_leased_notification_is_not_claimed_twice(self): + import data_loader_pg + from platform_notifications import _claim + + key = self._fresh_notification() + first = data_loader_pg.get_pg_worker_connection() + second = data_loader_pg.get_pg_worker_connection() + try: + with first.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur: + self.assertEqual([row["dedupe_key"] for row in _claim(cur, 5)], [key]) + first.commit() + with second.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur: + self.assertEqual(_claim(cur, 5), []) + second.commit() + finally: + first.close() + second.close() + + def test_decision_email_goes_to_requester_only(self): + from platform_notifications import deliver_pending_notifications + from role_requests import create_role_request_pg, decide_role_request_pg + + requester = _user(f"mail-{uuid.uuid4().hex[:6]}", "researcher") + self._owner_sql( + "INSERT INTO users (id, name, email, role, organization, status) VALUES (%s, %s, %s, 'researcher', %s, 'active')", + (requester.id, requester.name, requester.email, requester.organization), + ) + request = create_role_request_pg(user=requester, requested_role="equipment_owner", justification="I run the tool") + decide_role_request_pg(admin_user=self.admin, request_id=request["id"], decision="approve") + self._owner_sql( + "UPDATE platform_notifications SET next_attempt_at = now() + interval '1 day' " + "WHERE completed_at IS NULL AND dedupe_key <> %s", + (f"role_change_decided:{request['id']}",), + ) + posted = [] + + def fake_urlopen(request_obj, timeout=0): + posted.append((request_obj.full_url, json.loads(request_obj.data))) + return _FakeResponse() + + env = {"ADMIN_TEAMS_WEBHOOK_URL": "https://teams.example.test/hook", "EMAIL_WEBHOOK_URL": "https://mail.example.test/hook"} + with patch.dict(os.environ, env), patch("platform_notifications.urllib.request.urlopen", side_effect=fake_urlopen): + deliver_pending_notifications() + self.assertEqual([url for url, _ in posted], ["https://mail.example.test/hook"]) + self.assertEqual(posted[0][1]["to"], [requester.email]) + self.assertIn("approved", posted[0][1]["subject"]) + + # ── review round 1 ────────────────────────────────────────────────── + + def test_stale_claim_cannot_overwrite_a_newer_attempt(self): + import data_loader_pg + from platform_notifications import _claim, _deliver_one, _record_outcome + + key = self._fresh_notification() + conn = data_loader_pg.get_pg_worker_connection() + try: + with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur: + stale = _claim(cur, 1)[0] + conn.commit() + self._owner_sql( + "UPDATE platform_notifications SET lease_expires_at = now() - interval '1 second' WHERE dedupe_key = %s", + (key,), + ) + with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur: + fresh = _claim(cur, 1)[0] + conn.commit() + self.assertNotEqual(stale["claim_token"], fresh["claim_token"]) + outcome = _deliver_one(stale, []) + self.assertFalse(_record_outcome(conn, stale, outcome)) + self.assertTrue(_record_outcome(conn, fresh, _deliver_one(fresh, []))) + finally: + conn.close() + + def test_partial_smtp_refusal_retries_only_refused_recipients(self): + from platform_notifications import deliver_pending_notifications + + key = self._fresh_notification() + batches = [] + + class FakeSMTP: + def __init__(self, host, port, timeout=0): + pass + + def __enter__(self): + return self + + def __exit__(self, *exc): + return False + + def starttls(self, context=None): + pass + + def send_message(self, message): + batches.append(message["To"]) + return {"b@example.test": (450, b"try later")} if len(batches) == 1 else {} + + env = {"SMTP_HOST": "smtp.example.test", "SMTP_FROM": "dt@example.test", + "ADMIN_NOTIFICATION_EMAILS": "a@example.test,b@example.test"} + with patch.dict(os.environ, env), patch("platform_notifications.smtplib.SMTP", FakeSMTP): + deliver_pending_notifications() + teams, email, _attempts, done, error, _ = self._state(key) + self.assertEqual((email, done), ("pending", False)) + self.assertIn("1 recipients refused", error) + self._owner_sql("UPDATE platform_notifications SET next_attempt_at = now() WHERE dedupe_key = %s", (key,)) + deliver_pending_notifications() + self.assertEqual(batches, ["a@example.test, b@example.test", "b@example.test"]) + self.assertEqual(self._state(key)[1], "sent") + + def test_invalid_smtp_security_never_connects(self): + from platform_notifications import deliver_pending_notifications + + key = self._fresh_notification() + env = {"SMTP_HOST": "smtp.example.test", "SMTP_FROM": "dt@example.test", + "SMTP_SECURITY": "start_tls", "SMTP_USERNAME": "u", "SMTP_PASSWORD": "p"} + with patch.dict(os.environ, env), patch("platform_notifications.smtplib.SMTP") as smtp: + deliver_pending_notifications() + smtp.assert_not_called() + self.assertEqual(self._state(key)[1], "pending") + self.assertIn("ValueError", self._state(key)[4]) + + def test_stored_errors_never_contain_webhook_secrets(self): + from platform_notifications import deliver_pending_notifications + + key = self._fresh_notification() + env = {"ADMIN_TEAMS_WEBHOOK_URL": "prod.example.test/workflows/x?sig=SUPERSECRET"} + with patch.dict(os.environ, env): + deliver_pending_notifications() + error = self._state(key)[4] + self.assertTrue(error.startswith("teams: ")) + self.assertNotIn("SUPERSECRET", error) + self.assertNotIn("example.test", error) + + def test_daily_role_request_limit(self): + from role_requests import RoleChangeError, cancel_role_request_pg, create_role_request_pg + + person = _user(f"lim-{uuid.uuid4().hex[:6]}", "researcher") + self._owner_sql( + "INSERT INTO users (id, name, email, role, organization, status) VALUES (%s, %s, %s, 'researcher', %s, 'active')", + (person.id, person.name, person.email, person.organization), + ) + for _ in range(3): + request = create_role_request_pg(user=person, requested_role="pi", justification="again") + cancel_role_request_pg(user=person, request_id=request["id"]) + with self.assertRaises(RoleChangeError) as caught: + create_role_request_pg(user=person, requested_role="pi", justification="once more") + self.assertEqual(caught.exception.status_code, 429) + + def test_role_edit_before_first_sign_in_keeps_the_sign_in_alert(self): + client = self._client() + email = f"edited-{self.suffix}@example.com" + client.post("/api/admin/users", headers=self._headers(self.admin), json={"email": email, "role": "researcher"}) + edited = client.patch( + f"/api/admin/users/edited-{self.suffix}/role", + headers=self._headers(self.admin), + json={"role": "pi", "reason": "pre-approved"}, + ) + self.assertEqual(edited.status_code, 200, edited.text) + self.assertEqual(edited.json()["status"], "invited") + client.get(f"/api/admin/users/{email}/role", headers={"X-System-Token": os.environ["DT_SYSTEM_TOKEN"]}) + self.assertEqual(len(self._notifications("user_first_sign_in", f"user_first_sign_in:edited-{self.suffix}")), 1) + + def test_role_sync_refuses_proxied_calls(self): + client = self._client() + email = f"victim-{self.suffix}@example.test" + response = client.get(f"/api/admin/users/{email}/role", headers=self._headers(self.researcher)) + self.assertEqual(response.status_code, 403) + self.assertEqual(self._owner_sql("SELECT count(*) FROM users WHERE email = %s", (email,))[0][0], 0) + + def test_merging_accounts_moves_requests_and_keeps_one_pending(self): + import data_loader_pg + from role_requests import create_role_request_pg + from user_identity import sync_canonical_user + + email = f"merge-{self.suffix}@example.test" + old = _user(f"old-{self.suffix}", "researcher") + self._owner_sql( + "INSERT INTO users (id, name, email, role, organization, status) VALUES (%s, 'Old', %s, 'researcher', 'P', 'active')", + (old.id, email), + ) + older = create_role_request_pg(user=_user(old.id, "researcher"), requested_role="pi", justification="old account") + conn = data_loader_pg.get_pg_worker_connection() + try: + with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur: + merged = sync_canonical_user(cur, email=email, requested_user_id=f"new-{self.suffix}", name="New", organization="P") + conn.commit() + finally: + conn.close() + rows = self._owner_sql("SELECT user_id, status FROM role_change_requests WHERE id = %s", (older["id"],)) + self.assertEqual(rows, [(merged["id"], "pending")]) + + def test_rapid_equipment_resubmissions_notify_once(self): + from metadata_pg import register_equipment_pg, update_equipment_pg + + name = f"Rapid Etcher {self.suffix}" + created = register_equipment_pg( + payload={"name": name, "columns": [{"name": "p", "type": "float"}], "primary_target": "rate"}, + user=self.owner, + ) + update_equipment_pg( + domain_id=created["domain_id"], + payload={"name": name, "columns": [{"name": "p", "type": "float"}, {"name": "q", "type": "float"}], "primary_target": "rate"}, + user=self.owner, + ) + rows = self._notifications("equipment_review_requested", f"equipment_review_requested:{created['registration_id']}:%") + self.assertEqual(len(rows), 1) + + def test_resubmission_after_approval_starts_a_new_review_alert(self): + from metadata_pg import register_equipment_pg, update_equipment_pg + + name = f"Cycle Etcher {self.suffix}" + payload = {"name": name, "columns": [{"name": "p", "type": "float"}], "primary_target": "rate"} + created = register_equipment_pg(payload=payload, user=self.owner) + self._owner_sql( + "UPDATE equipment_registrations SET status = 'approved' WHERE domain_id = %s", (created["domain_id"],) + ) + self._owner_sql( + "UPDATE equipment_metadata SET status = 'approved' WHERE domain_id = %s", (created["domain_id"],) + ) + revised = update_equipment_pg( + domain_id=created["domain_id"], + payload={**payload, "columns": [{"name": "p", "type": "float"}, {"name": "r", "type": "float"}]}, + user=self.owner, + ) + self.assertEqual(revised["status"], "pending") + rows = self._notifications("equipment_review_requested", f"equipment_review_requested:{created['registration_id']}:%") + self.assertEqual(len(rows), 2) + + def test_subjects_are_single_line(self): + import data_loader_pg + from platform_notifications import enqueue_notification + + key = f"test:{uuid.uuid4().hex}" + conn = data_loader_pg.get_pg_worker_connection() + try: + with conn.cursor() as cur: + enqueue_notification(cur, kind="test", dedupe_key=key, subject="Evil\r\nBcc: x@example.test", message="m") + conn.commit() + finally: + conn.close() + self.assertEqual( + self._owner_sql("SELECT subject FROM platform_notifications WHERE dedupe_key = %s", (key,))[0][0], + "Evil Bcc: x@example.test", + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/api/tests/test_platform_notifications.py b/api/tests/test_platform_notifications.py new file mode 100644 index 0000000..3621658 --- /dev/null +++ b/api/tests/test_platform_notifications.py @@ -0,0 +1,132 @@ +from __future__ import annotations + +import os +import sys +import unittest +from pathlib import Path +from unittest.mock import patch + +API_ROOT = Path(__file__).resolve().parents[1] +if str(API_ROOT) not in sys.path: + sys.path.insert(0, str(API_ROOT)) + +os.environ.setdefault("DT_SYSTEM_TOKEN", "test-system-token") +os.environ.setdefault("INGESTION_TOKEN", "test-ingestion-token") +os.environ.setdefault("PG_PASS", "test-api-password") +os.environ.setdefault("PG_SUPER_PASS", "test-super-password") + +import platform_notifications as pn # noqa: E402 +import role_requests # noqa: E402 +from security import PlatformUser # noqa: E402 + +CLEAR_ENV = {key: "" for key in ( + "ADMIN_TEAMS_WEBHOOK_URL", "EMAIL_WEBHOOK_URL", "SMTP_HOST", "PLATFORM_BASE_URL", + "ADMIN_NOTIFICATION_EMAILS", "NOTIFICATION_MAX_ATTEMPTS", +)} + + +def _notification(**overrides): + base = { + "id": "n1", + "kind": "test", + "subject": "Role request: