diff --git a/scripts/browser_journeys.mjs b/scripts/browser_journeys.mjs index bd968fd..eb1d7f4 100644 --- a/scripts/browser_journeys.mjs +++ b/scripts/browser_journeys.mjs @@ -41,7 +41,20 @@ try{ await identity('operator');await navigate('/platform'); await check(`!!document.querySelector('a[href="/platform/operations"]')`,'P01 platform recovery navigation'); await navigate('/platform/operations'); - await check(`document.body.innerText.includes("Live sign-in, email receipt and authenticator health are not verified here")`,'P05 unknown provider health remains explicit'); + await check(`document.body.innerText.includes("Component checks do not prove a complete sign-in or inbox receipt")`,'P05 unknown provider health remains explicit'); + await navigate('/platform/operations?event_id=p05-browser'); + await check(`document.body.innerText.includes("No mail submission is recorded")`,'P05 selected mail evidence'); + await evaluate(`document.querySelector('form[action="/platform/operations/deliver"] button').click()`); + await waitFor('document.body.innerText.includes("Confirm change")'); + await check(`document.body.innerText.includes("p05-browser")`,'P05 selected delivery requires review'); + await evaluate(`document.querySelector('input[name="confirm_token"]').form.querySelector('button').click()`); + await waitFor('document.body.innerText.includes("definitely failed")'); + await check(`!!document.querySelector('form[action="/platform/operations/deliver"]')`,'P05 failure offers controlled retry'); + await evaluate(`document.querySelector('form[action="/platform/operations/deliver"] button').click()`); + await waitFor('document.body.innerText.includes("Confirm change")'); + await evaluate(`document.querySelector('input[name="confirm_token"]').form.querySelector('button').click()`); + await waitFor('document.body.innerText.includes("mail provider accepted")'); + await check(`!document.querySelector('form[action="/platform/operations/deliver"]') && document.body.innerText.includes("does not prove inbox receipt")`,'P05 completed delivery cannot be blindly repeated'); await navigate('/platform/activity'); await check(`!!document.querySelector('input[name="reference"]') && !!document.querySelector('input[name="tenant"]')`,'P08 platform investigation filters'); await evaluate(`document.querySelector('input[name="reference"]').value='synthetic-missing';document.querySelector('form[action="/platform/activity"] button').click()`); diff --git a/scripts/browser_journeys.py b/scripts/browser_journeys.py index 797d626..80ddb0d 100644 --- a/scripts/browser_journeys.py +++ b/scripts/browser_journeys.py @@ -22,6 +22,16 @@ if not chrome: if not chrome or not shutil.which('node'): raise SystemExit('Chromium and Node are required; set JOURNEY_CHROME to the Chromium executable. Browser tests were not run.') fixture=FactorRecoveryJourney();fixture.setUp();fixture.member(email='actual.login@example.test') +from user_engine.domain import OutboxEvent +fixture.app.service.store.append_outbox(OutboxEvent(event_id="p05-browser", event_type="family_member.invited", aggregate_id="fixture", tenant="tenant:trial:demo-company", correlation_id="p05-browser", payload={"primary_email":"fixture@example.test", "invitation_id":"fixture"})) +class Delivery: + attempts = 0 + def __call__(self, event): + self.attempts += 1 + if self.attempts == 1: raise RuntimeError("synthetic outage") + def mail_delivery_status(self, event_id): + return {"state":"failed" if self.attempts == 1 else "provider_accepted" if self.attempts else "not_found"} +fixture.app.outbox_delivery = Delivery() class Quiet(WSGIRequestHandler): def log_message(self,*args):pass server=make_server('127.0.0.1',0,fixture.app,handler_class=Quiet) diff --git a/scripts/p05_mail_acceptance.py b/scripts/p05_mail_acceptance.py new file mode 100644 index 0000000..a70d181 --- /dev/null +++ b/scripts/p05_mail_acceptance.py @@ -0,0 +1,123 @@ +#!/usr/bin/env python3 +"""P05 portal -> HTTP -> SMTP/IMAP acceptance; loopback fixtures only. + +Requires sibling email-connect source and its owned GreenMail harness on 3025/3143. +No production configuration or recipients are loaded. +""" + +import imaplib +import json +from pathlib import Path +import sys +import tempfile +from threading import Thread +from wsgiref.simple_server import make_server, WSGIRequestHandler + +ROOT = Path(__file__).resolve().parents[1] +sys.path[:0] = [ + str(ROOT / "src"), + str(ROOT / "tests"), + str(ROOT.parent / "email-connect/src"), +] +from email_connect.transactional import ( + SMTPProvider, + SQLiteDeliveryStore, + TransactionalApplication, +) +from user_engine.adapters.delivery import HTTPOutboxDeliveryAdapter +from test_service_operations import ServiceOperations + + +class Quiet(WSGIRequestHandler): + def log_message(self, *args): + pass + + +with tempfile.TemporaryDirectory(prefix="p05-mail-") as directory: + mail = TransactionalApplication( + SQLiteDeliveryStore(directory + "/mail.db"), + SMTPProvider( + "127.0.0.1", + 3025, + "fixture", + "fixture", + "noreply@harness.email-connect.test", + security="plaintext", + ), + "fixture-bearer", + "https://users.example.test", + ) + calls = [] + + def receiver(env, start): + if env["PATH_INFO"] == "/audit": + calls.append(True) + start( + "503 Service Unavailable" if len(calls) == 1 else "202 Accepted", + [("Content-Type", "application/json")], + ) + return [b"{}"] + return mail(env, start) + + server = make_server("127.0.0.1", 0, receiver, handler_class=Quiet) + thread = Thread(target=server.serve_forever, daemon=True) + thread.start() + fixture = ServiceOperations() + fixture.setUp() + try: + event = fixture.seed() + address = event.payload["primary_email"] + with imaplib.IMAP4("127.0.0.1", 3143) as inbox: + inbox.login(address, "fixture") + inbox.select("INBOX") + before = len(inbox.search(None, "ALL")[1][0].split()) + base = "http://127.0.0.1:" + str(server.server_port) + adapter = HTTPOutboxDeliveryAdapter( + event_url=base + "/audit", + event_bearer_token="fixture-bearer", + mail_url=base + "/v1/send", + mail_bearer_token="fixture-bearer", + ) + fixture.app.outbox_delivery = adapter + assert adapter.mail_status()["checks"]["smtp_authentication"] == "ok" + fixture.post( + "/platform/operations/deliver", + who="operator", + event_id=event.event_id, + confirmed=True, + ) + first = fixture.app.service.store.outbox_event(event.event_id) + assert ( + first.failed_at and first.claimed_by is None and first.delivered_at is None + ) + assert ( + adapter.mail_delivery_status(event.event_id)["state"] == "provider_accepted" + ) + fixture.post( + "/platform/operations/deliver", + who="operator", + event_id=event.event_id, + confirmed=True, + ) + assert fixture.app.service.store.outbox_event(event.event_id).delivered_at + with imaplib.IMAP4("127.0.0.1", 3143) as inbox: + inbox.login(address, "fixture") + inbox.select("INBOX") + after = len(inbox.search(None, "ALL")[1][0].split()) + assert after - before == 1 and len(calls) == 2 + print( + json.dumps( + { + "success": True, + "scope": "loopback SMTP/IMAP fixtures only", + "mail_received": 1, + "audit_failure_recovered": True, + "retry_did_not_duplicate_mail": True, + } + ) + ) + finally: + server.shutdown() + server.server_close() + thread.join(timeout=5) + fixture.tearDown() diff --git a/src/user_engine/adapters/delivery.py b/src/user_engine/adapters/delivery.py index a532a38..6c8bdc0 100644 --- a/src/user_engine/adapters/delivery.py +++ b/src/user_engine/adapters/delivery.py @@ -3,7 +3,20 @@ from __future__ import annotations import json -from urllib.request import Request, urlopen +from urllib.request import Request, urlopen, build_opener, HTTPRedirectHandler +from urllib.error import HTTPError + + +class DeliveryError(RuntimeError): + def __init__(self, code): + self.code = code + super().__init__(code) + + +class NoRedirect(HTTPRedirectHandler): + def redirect_request(self, *args, **kwargs): + return None + from user_engine.domain import OutboxEvent @@ -39,6 +52,8 @@ class HTTPOutboxDeliveryAdapter: "occurred_at": event.occurred_at.isoformat(), "data": dict(event.payload), } + if event.event_type in _MAIL_EVENTS and not self.mail_url: + raise DeliveryError("mail_unconfigured") if self.mail_url and event.event_type in _MAIL_EVENTS: if not self.mail_bearer_token: raise RuntimeError("mail delivery token is required") @@ -51,21 +66,78 @@ class HTTPOutboxDeliveryAdapter: event_envelope["data"] = event_data self._post(self.event_url, event_envelope, self.event_bearer_token) - def _post( - self, url: str, envelope: dict[str, object], bearer_token: str - ) -> None: - with urlopen( - Request( - url, - data=json.dumps(envelope).encode(), - headers={ - "Authorization": f"Bearer {bearer_token}", - "Content-Type": "application/json", - "Idempotency-Key": str(envelope["id"]), - }, - method="POST", - ), - timeout=self.timeout_seconds, + def _post(self, url: str, envelope: dict[str, object], bearer_token: str) -> None: + request = Request( + url, + data=json.dumps(envelope).encode(), + headers={ + "Authorization": f"Bearer {bearer_token}", + "Content-Type": "application/json", + "Idempotency-Key": str(envelope["id"]), + }, + method="POST", + ) + try: + response = build_opener(NoRedirect()).open( + request, timeout=self.timeout_seconds + ) + except HTTPError as exc: + code = "provider_unavailable" + try: + body = json.loads(exc.read(8192)) + candidate = body.get("error") + if candidate in { + "delivery_outcome_unknown", + "temporary_deferral", + "permanent_rejection", + "recipient_suppressed", + }: + code = candidate + except Exception: + pass + raise DeliveryError(code) from None + except Exception: + raise DeliveryError("provider_unavailable") from None + with response: + if not 200 <= response.status < 300: + raise DeliveryError("provider_unavailable") + + def mail_status(self): + if not self.mail_url or not self.mail_bearer_token: + return {"checks": {"smtp_authentication": "unconfigured"}} + from urllib.parse import urlsplit, urlunsplit + + parsed = urlsplit(self.mail_url) + url = urlunsplit((parsed.scheme, parsed.netloc, "/v1/status", "", "")) + req = Request( + url, + data=b"{}", + headers={ + "Authorization": "Bearer " + self.mail_bearer_token, + "Content-Type": "application/json", + }, + method="POST", + ) + with build_opener(NoRedirect()).open(req, timeout=7) as response: + return json.loads(response.read(8192)) + + def mail_delivery_status(self, event_id): + if not self.mail_url or not self.mail_bearer_token: + return {"state": "unconfigured"} + from urllib.parse import urlsplit, urlunsplit + + parsed = urlsplit(self.mail_url) + url = urlunsplit((parsed.scheme, parsed.netloc, "/v1/delivery-status", "", "")) + request = Request( + url, + data=json.dumps({"event_id": event_id}).encode(), + headers={ + "Authorization": "Bearer " + self.mail_bearer_token, + "Content-Type": "application/json", + }, + method="POST", + ) + with build_opener(NoRedirect()).open( + request, timeout=self.timeout_seconds ) as response: - if response.status < 200 or response.status >= 300: - raise RuntimeError(f"delivery rejected with HTTP {response.status}") + return json.loads(response.read(8192)) diff --git a/src/user_engine/adapters/local.py b/src/user_engine/adapters/local.py index 927d652..c5290ba 100644 --- a/src/user_engine/adapters/local.py +++ b/src/user_engine/adapters/local.py @@ -87,6 +87,12 @@ class InMemoryUserEngineStore: with self._lifecycle_lock: yield + @contextmanager + def outbox_delivery_guard(self, event_id: str): + # Separate namespace; reuse the existing cross-connection lock primitive. + with self.tenant_lifecycle_guard("outbox-event:" + event_id): + yield + def outbox_history(self) -> tuple[OutboxEvent, ...]: return tuple(self.outbox_events) diff --git a/src/user_engine/adapters/postgres.py b/src/user_engine/adapters/postgres.py index 075946e..5325d09 100644 --- a/src/user_engine/adapters/postgres.py +++ b/src/user_engine/adapters/postgres.py @@ -103,6 +103,12 @@ class PostgresUserEngineStore: with self._cursor() as cursor: cursor.execute("SELECT pg_advisory_unlock(hashtextextended(%s, 0))", (key,)) + @contextmanager + def outbox_delivery_guard(self, event_id: str): + # Separate namespace; reuse the existing cross-connection lock primitive. + with self.tenant_lifecycle_guard("outbox-event:" + event_id): + yield + def outbox_history(self) -> tuple[OutboxEvent, ...]: with self._cursor() as cursor: cursor.execute("SELECT payload FROM user_engine_outbox_events ORDER BY occurred_at, event_id") diff --git a/src/user_engine/operations_status.py b/src/user_engine/operations_status.py new file mode 100644 index 0000000..956a9d0 --- /dev/null +++ b/src/user_engine/operations_status.py @@ -0,0 +1,143 @@ +"""Bounded, non-sending diagnostics rendered only after platform authorization.""" + +from concurrent.futures import ThreadPoolExecutor +from datetime import datetime, timezone +from html import escape +import json +from urllib.request import Request, build_opener, HTTPRedirectHandler +from urllib.error import HTTPError + + +class NoRedirect(HTTPRedirectHandler): + def redirect_request(self, *args, **kwargs): + return None + + +def sign_in_status(base): + try: + response = build_opener(NoRedirect()).open( + base.rstrip("/") + "/readyz", timeout=5 + ) + except HTTPError as exc: + response = exc + with response: + raw = response.read(16385) + if len(raw) > 16384: + raise ValueError("oversized status") + return json.loads(raw) + + +def check_services(oidc, delivery): + def safe(call): + try: + return call() + except Exception: + return {} + + with ThreadPoolExecutor(max_workers=2) as pool: + sign = pool.submit( + safe, lambda: sign_in_status(oidc.backend_url) if oidc else {} + ) + mail = pool.submit( + safe, + lambda: delivery.mail_status() if hasattr(delivery, "mail_status") else {}, + ) + sign, mail = sign.result(), mail.result() + values = {} + sign_checks = sign.get("checks", []) if isinstance(sign, dict) else [] + for row in sign_checks if isinstance(sign_checks, list) else []: + if isinstance(row, dict): + values[row.get("name")] = row.get("status") + checks = [ + ( + "Directory sign-in", + values.get("lldap"), + "Sign-in and identity lookup", + "Retry the check; if it still fails, ask the identity service owner to restore directory connectivity.", + ), + ( + "Primary authentication", + values.get("authelia"), + "Fresh sign-in", + "Retry after the authentication service recovers. Existing sessions do not prove new sign-in works.", + ), + ( + "Authenticator provider", + values.get("privacyidea"), + "MFA verification and enrollment", + "Retry after the MFA service recovers. Do not disable MFA to work around an outage.", + ), + ( + "Factor-read credential", + values.get("factor_reader"), + "Enrollment decisions and MFA verification", + "Renewal is automatic. A failure can mean expiry, lost permissions or projection failure; ask the credential owner to run the scoped renewal check.", + ), + ] + statuses = mail.get("checks", {}) if isinstance(mail, dict) else {} + if not isinstance(statuses, dict): + statuses = {} + checks += [ + ( + "Mail record store", + statuses.get("database"), + "Durable delivery tracking", + "Restore the mail store before resuming delivery.", + ), + ( + "SMTP authentication", + statuses.get("smtp_authentication"), + "Invitation and verification mail", + "Check the configured mail lane. Use assisted password setup when the person cannot receive email.", + ), + ] + rows = "" + for name, status, impact, recovery in checks: + label = { + "ok": "Check passed", + "failed": "Unavailable", + "unconfigured": "Not configured", + }.get(status, "Unknown — check unavailable") + rows += ( + "" + + escape(name) + + "" + + escape(label) + + "" + + escape(impact) + + "" + + escape(recovery) + + "" + ) + return ( + "

Checked service state

Checked at " + + escape(datetime.now(timezone.utc).isoformat(timespec="seconds")) + + ". SMTP authentication may be cached for 30 seconds. These checks send no email and do not prove inbox receipt.

" + + rows + + '
ComponentStateAffected journeysRecovery

Check services again

' + ) + + +def selected_mail_status(delivery, event): + if event.event_type not in {"family_member.invited", "family_invitation.resent"}: + return "" + try: + receipt = delivery.mail_delivery_status(event.event_id) + state = receipt.get("state") if isinstance(receipt, dict) else None + except Exception: + state = None + detail = { + "provider_accepted": "The mail provider accepted this event. This does not prove inbox receipt. Retrying this same record reconciles the remaining event delivery without submitting another invitation.", + "unknown": "The SMTP outcome is uncertain. Automatic resending is blocked. Ask the mail service owner to reconcile the provider record using this delivery ID; use assisted password setup if access is urgent.", + "failed": "The previous mail attempt definitely failed. Review a delivery attempt to retry this same record.", + "not_found": "No mail submission is recorded. Review a delivery attempt to submit this selected record.", + "unconfigured": "The mail lane is not configured. Use assisted password setup or restore the mail lane.", + }.get( + state, + "Mail evidence is unavailable. Check the mail service and retry this status check; do not infer that a message was sent or received.", + ) + return ( + "

Selected mail evidence

" + + escape(detail) + + "

" + ) diff --git a/src/user_engine/ports.py b/src/user_engine/ports.py index 67b8565..bb8d783 100644 --- a/src/user_engine/ports.py +++ b/src/user_engine/ports.py @@ -437,6 +437,9 @@ class UserEngineStore(Protocol): def pending_outbox(self) -> tuple[OutboxEvent, ...]: """Return pending outbox events in write order.""" + def outbox_delivery_guard(self, event_id: str) -> ContextManager[None]: + """Serialize delivery/replay for one event across processes.""" + def save_outbox(self, event: OutboxEvent) -> None: """Persist outbox delivery state.""" diff --git a/src/user_engine/runtime.py b/src/user_engine/runtime.py index 68279b2..e13f941 100644 --- a/src/user_engine/runtime.py +++ b/src/user_engine/runtime.py @@ -63,7 +63,7 @@ def create_application() -> PortalApplication: ), timeout_seconds=float(os.environ.get("USER_ENGINE_DELIVERY_TIMEOUT", "5")), ) - return PortalApplication( + app = PortalApplication( service, trusted_proxy_secret=_required("USER_ENGINE_PROXY_SECRET"), login_url=_required("USER_ENGINE_LOGIN_URL"), @@ -120,6 +120,9 @@ def create_application() -> PortalApplication: ), ) + from user_engine.operations_status import check_services + app.operations_probe = lambda: check_services(app.oidc_client, app.outbox_delivery) + return app def main() -> None: host = os.environ.get("USER_ENGINE_HOST", "0.0.0.0") diff --git a/src/user_engine/service.py b/src/user_engine/service.py index a945fa3..d9193a0 100644 --- a/src/user_engine/service.py +++ b/src/user_engine/service.py @@ -2867,71 +2867,65 @@ class UserEngineService: return self.store.pending_outbox() def deliver_outbox( - self, - actor: Actor, - deliver: Callable[[OutboxEvent], None], - *, - worker_id: str, - max_attempts: int = 3, - correlation_id: str | None = None, + self, actor: Actor, deliver: Callable[[OutboxEvent], None], *, + worker_id: str, max_attempts: int = 3, correlation_id: str | None = None, + event_id: str | None = None, expected_attempts: int | None = None, ) -> tuple[OutboxEvent, ...]: - """Claim and deliver pending events, retaining bounded failure details.""" - if not worker_id.strip(): - raise ValidationError("outbox worker_id is required") - if not 1 <= max_attempts <= 20: - raise ValidationError("outbox max_attempts must be between 1 and 20") + """Serialize each external attempt and retain only classified failures.""" + if not worker_id.strip() or not 1 <= max_attempts <= 20: + raise ValidationError("invalid outbox worker or attempt limit") correlation_id = correlation_id or new_id("corr") - self._authorize( - actor, action="outbox.deliver", resource_type="user-engine:outbox", - resource_id="pending", tenant=actor.tenant, - correlation_id=correlation_id, - ) - results: list[OutboxEvent] = [] - for event in self.store.pending_outbox(): - claimed = replace( - event, claimed_by=worker_id, claimed_at=utc_now(), - delivery_attempts=event.delivery_attempts + 1, - failed_at=None, failure_reason=None, - ) - with self.store.transaction(): - self.store.save_outbox(claimed) - try: - deliver(claimed) - except Exception as exc: - reason = str(exc).strip()[:200] or type(exc).__name__ - failed = replace( - claimed, failed_at=utc_now(), failure_reason=reason, - dead_lettered_at=(utc_now() if claimed.delivery_attempts >= max_attempts else None), - ) - with self.store.transaction(): - self.store.save_outbox(failed) - results.append(failed) - else: - delivered = replace(claimed, delivered_at=utc_now()) - with self.store.transaction(): - self.store.save_outbox(delivered) - results.append(delivered) + self._authorize(actor, action="outbox.deliver", resource_type="user-engine:outbox", + resource_id="pending", tenant=actor.tenant, correlation_id=correlation_id) + selected = self.store.outbox_event(event_id) if event_id else None + if event_id and selected is None: + raise NotFoundError("delivery record not found") + events = (selected,) if selected else self.store.pending_outbox() + results = [] + for candidate in events: + with self.store.outbox_delivery_guard(candidate.event_id): + event = self.store.outbox_event(candidate.event_id) + if event is None or event.delivered_at is not None: + continue + if expected_attempts is not None and event.delivery_attempts != expected_attempts: + raise ConflictError("Delivery changed. Refresh and review the attempt again.") + # Owning this cross-process guard proves any previous claim is orphaned. + # Receiver idempotency reconciles a lost receipt without blind resending. + if event.dead_lettered_at and not event_id: + continue + claimed = replace(event, claimed_by=worker_id, claimed_at=utc_now(), + delivery_attempts=event.delivery_attempts+1, failed_at=None, failure_reason=None) + with self.store.transaction(): self.store.save_outbox(claimed) + try: + deliver(claimed) + except Exception as exc: + code = getattr(exc, "code", "delivery_unavailable") + if code not in {"mail_unconfigured", "delivery_outcome_unknown", "temporary_deferral", "permanent_rejection", "recipient_suppressed", "provider_timeout", "provider_unavailable"}: + code = "delivery_unavailable" + outcome = replace(claimed, claimed_by=None, claimed_at=None, + failed_at=utc_now(), failure_reason=code, + dead_lettered_at=utc_now() if claimed.delivery_attempts >= max_attempts else None) + else: + outcome = replace(claimed, claimed_by=None, claimed_at=None, + delivered_at=utc_now(), dead_lettered_at=None) + with self.store.transaction(): self.store.save_outbox(outcome) + results.append(outcome) return tuple(results) - def replay_outbox( - self, actor: Actor, event_id: str, *, correlation_id: str | None = None - ) -> OutboxEvent: + def replay_outbox(self, actor: Actor, event_id: str, *, correlation_id: str | None = None) -> OutboxEvent: correlation_id = correlation_id or new_id("corr") - event = self.store.outbox_event(event_id) - if event is None: - raise NotFoundError("outbox event not found") - self.resolve_tenant_context(actor, event.tenant) - self._authorize( - actor, action="outbox.replay", resource_type="user-engine:outbox-event", - resource_id=event_id, tenant=event.tenant, correlation_id=correlation_id, - ) - replayed = replace( - event, claimed_by=None, claimed_at=None, failed_at=None, - failure_reason=None, dead_lettered_at=None, - ) - with self.store.transaction(): - self.store.save_outbox(replayed) - return replayed + with self.store.outbox_delivery_guard(event_id): + event = self.store.outbox_event(event_id) + if event is None: raise NotFoundError("outbox event not found") + self.resolve_tenant_context(actor, event.tenant) + self._authorize(actor, action="outbox.replay", resource_type="user-engine:outbox-event", + resource_id=event_id, tenant=event.tenant, correlation_id=correlation_id) + if event.delivered_at or (event.claimed_by and not event.failed_at): + raise ConflictError("Delivery is completed or its previous attempt is unresolved.") + replayed = replace(event, claimed_by=None, claimed_at=None, failed_at=None, + failure_reason=None, dead_lettered_at=None) + with self.store.transaction(): self.store.save_outbox(replayed) + return replayed def outbox_diagnostics(self) -> OutboxDiagnostics: event_types: dict[str, int] = {} diff --git a/src/user_engine/web.py b/src/user_engine/web.py index f1af186..fc9557c 100644 --- a/src/user_engine/web.py +++ b/src/user_engine/web.py @@ -109,6 +109,7 @@ class PortalApplication: self.provisioning = provisioning self.tenant_management = tenant_management self.outbox_delivery = outbox_delivery + self.operations_probe = None self.registration_verification = registration_verification self.registration_clients = frozenset(registration_clients) self.registration_tenants = frozenset(registration_tenants) @@ -791,6 +792,27 @@ class PortalApplication: query = parse_qs(str(environ.get("QUERY_STRING", ""))) return self._html(start_response, self._platform_activity( query.get("reference", [""])[0], query.get("tenant", [""])[0]), correlation_id) + if path == "/platform/operations/deliver" and method == "POST": + self.service.resolve_tenant_context(actor, PLATFORM_TENANT) + body = self._form_body(environ) + self._require_csrf(environ, body.get("csrf_token", "")) + event_id = body.get("event_id", "") + with self.service.store.outbox_delivery_guard(event_id): + event = self.service.store.outbox_event(event_id) + if event is None: raise NotFoundError("delivery record not found") + if event.delivered_at: + raise ConflictError("Delivery is already completed.") + if self.outbox_delivery is None: raise ValidationError("Delivery is unavailable") + confirmation = self._confirm_change(environ, start_response, body, + json.dumps(_jsonable(event),sort_keys=True), "Attempt this delivery", + "This submits only delivery " + event.event_id + " for " + event.tenant + + ". Invitation messages will be sent to the recorded recipient. Provider acceptance does not prove the person received it. Keep support reference " + event.correlation_id + ".", + correlation_id) + if confirmation is not None: return confirmation + self.service.deliver_outbox(actor, self.outbox_delivery, + worker_id="platform-operator",event_id=event.event_id, + expected_attempts=event.delivery_attempts,correlation_id=correlation_id) + return self._redirect(start_response,"/platform/operations?"+urlencode({"event_id":event_id}),correlation_id) if path in {"/platform/operations", "/platform/operations/replay"}: self.service.resolve_tenant_context(actor, PLATFORM_TENANT) if method == "POST" and path.endswith("/replay"): @@ -799,7 +821,7 @@ class PortalApplication: event = self.service.store.outbox_event(str(body.get("event_id", ""))) if event is None: raise NotFoundError("delivery record not found") - if event.delivered_at is not None or event.claimed_by: + if event.delivered_at is not None or (event.claimed_by and not event.failed_at): raise ConflictError("Delivery is already completed or being processed. Refresh its status.") self.service.replay_outbox(actor, event.event_id, correlation_id=correlation_id) return self._redirect(start_response, "/platform/operations?"+urlencode({"event_id":event.event_id}), correlation_id) @@ -1191,6 +1213,8 @@ class PortalApplication: @staticmethod def _delivery_status(event: Any) -> str: if event.delivered_at: return "Accepted by delivery adapter; receipt by the person is unverified" + if event.failure_reason == "delivery_outcome_unknown": return "Mail outcome unknown — check provider evidence before sending another message" + if event.failure_reason == "mail_unconfigured": return "Mail is not configured — use assisted setup or restore the mail lane" if event.dead_lettered_at: return "Delivery stopped after repeated failures" if event.failed_at: return "Delivery failed; retry pending" if event.claimed_by: return "Being processed" @@ -1242,15 +1266,19 @@ class PortalApplication: rows = "" for event in events: action = "" - if event.delivered_at is None and not event.claimed_by and (event.failed_at or event.dead_lettered_at): - action = f'
' + if event.delivered_at is None: + action = f'
' + if event.delivered_at is None and (not event.claimed_by or event.failed_at) and (event.failed_at or event.dead_lettered_at): + action += f'
' rows += f'{escape(event.event_id)}{escape(event.tenant)}{escape(event.event_type)}{escape(self._delivery_status(event))}{escape(event.correlation_id)}{action}' - configuration = self._operation_capabilities() - return self._page_html("Service recovery", '

Service recovery

' + configuration + '

Investigate a support reference

This view shows local delivery records. Live sign-in, email receipt and authenticator health are not verified here.

' + from user_engine.operations_status import check_services, selected_mail_status + evidence = selected_mail_status(self.outbox_delivery, events[0]) if event_id else "" + configuration = (self.operations_probe() if self.operations_probe else check_services(None,None)) + self._operation_capabilities() + return self._page_html("Service recovery", '

Service recovery

' + configuration + evidence + '

Investigate a support reference

Delivery records below describe recorded attempts. Component checks do not prove a complete sign-in or inbox receipt.

' '
' '' + (rows or '') - + '
DeliveryTenantKindStatusSupport referenceRecovery
No delivery records. This does not prove mail was received.

Queued retries are processed by the delivery worker. Check the record again for the result.

Return to platform administration

') + + '

Queueing a retry does not send it. Use Review delivery attempt to submit one selected record, then check its result. No background worker is enabled here.

Return to platform administration

') def _operation_capabilities(self) -> str: capabilities = ( diff --git a/tests/test_journey_postgres.py b/tests/test_journey_postgres.py index 0d41b1a..186e414 100644 --- a/tests/test_journey_postgres.py +++ b/tests/test_journey_postgres.py @@ -92,3 +92,16 @@ class PostgresJourneyTests(unittest.TestCase): self.assertIsNone(second.user(u.user_id)) self.seed.commit() self.assertEqual(u,second.user(u.user_id)) + + def test_two_connections_deliver_one_event_once(self): + from user_engine.domain import OutboxEvent + event=OutboxEvent(event_id='p05-concurrent',event_type='membership.added',aggregate_id='fixture',tenant=self.tenant,correlation_id='p05',payload={}) + service=self.service(self.seed) + with service.store.transaction():service.store.append_outbox(event) + services=[self.service(self.connect()),self.service(self.connect())] + barrier=Barrier(2);calls=[] + def run(service): + barrier.wait(timeout=5) + return service.deliver_outbox(self.actor,lambda e:calls.append(e.event_id),worker_id='fixture',event_id=event.event_id) + with ThreadPoolExecutor(max_workers=2) as pool:list(pool.map(run,services)) + self.assertEqual(['p05-concurrent'],calls) diff --git a/tests/test_platform_adapters.py b/tests/test_platform_adapters.py index 0103533..eaef9b1 100644 --- a/tests/test_platform_adapters.py +++ b/tests/test_platform_adapters.py @@ -118,7 +118,9 @@ class PlatformAdapterTests(unittest.TestCase): event_url="http://events", mail_url="http://mail", event_bearer_token="event-opaque", mail_bearer_token="mail-opaque", ) - with patch("user_engine.adapters.delivery.urlopen", return_value=_Response()) as call: + with patch("user_engine.adapters.delivery.build_opener") as factory: + call=factory.return_value.open + call.return_value=_Response() adapter(_event("family_member.invited")) self.assertEqual([item.args[0].full_url for item in call.call_args_list], ["http://mail", "http://events"]) @@ -139,7 +141,9 @@ class PlatformAdapterTests(unittest.TestCase): event_url="http://events", mail_url="http://mail", event_bearer_token="event-opaque", mail_bearer_token="mail-opaque", ) - with patch("user_engine.adapters.delivery.urlopen", return_value=_Response()) as call: + with patch("user_engine.adapters.delivery.build_opener") as factory: + call=factory.return_value.open + call.return_value=_Response() adapter(_event("membership.added")) self.assertEqual(call.call_count, 1) self.assertEqual(call.call_args.args[0].full_url, "http://events") diff --git a/tests/test_service_operations.py b/tests/test_service_operations.py new file mode 100644 index 0000000..47b875f --- /dev/null +++ b/tests/test_service_operations.py @@ -0,0 +1,172 @@ +import time +from unittest.mock import patch +from dataclasses import replace +from user_engine.domain import OutboxEvent, utc_now +from user_engine.adapters.delivery import HTTPOutboxDeliveryAdapter, DeliveryError +from user_engine.operations_status import check_services +from test_journey_roles import JourneyFixture, TENANT +from test_web import invoke + + +class ServiceOperations(JourneyFixture): + def seed(self): + event = OutboxEvent( + event_id="p05-one", + event_type="family_member.invited", + aggregate_id="person", + tenant=TENANT, + correlation_id="p05-case", + payload={ + "primary_email": "fixture@example.test", + "invitation_id": "fixture", + }, + ) + self.app.service.store.append_outbox(event) + self.calls = [] + self.app.outbox_delivery = lambda event: self.calls.append(event.event_id) + return event + + def test_one_confirmed_delivery_and_completed_retry_guard(self): + event = self.seed() + self.app.service.store.append_outbox(replace(event, event_id="p05-other")) + response, body = self.post( + "/platform/operations/deliver", who="operator", event_id=event.event_id + ) + self.assertIn(b"Confirm change", body) + self.assertEqual([], self.calls) + response, _ = self.post( + "/platform/operations/deliver", + who="operator", + event_id=event.event_id, + confirmed=True, + ) + self.assertEqual("303 See Other", response["status"]) + self.assertEqual(["p05-one"], self.calls) + response, _ = self.post( + "/platform/operations/deliver", who="operator", event_id=event.event_id + ) + self.assertEqual("409 Conflict", response["status"]) + self.assertIsNone(self.app.service.store.outbox_event("p05-other").delivered_at) + + def test_role_csrf_and_stale_confirmation_cannot_submit(self): + event = self.seed() + for who in ["member", "admin"]: + response, _ = self.post( + "/platform/operations/deliver", who=who, event_id=event.event_id + ) + self.assertEqual("403 Forbidden", response["status"]) + response, _ = self.post( + "/platform/operations/deliver", + who="operator", + event_id=event.event_id, + csrf_token="wrong", + ) + self.assertEqual("403 Forbidden", response["status"]) + _, body = self.post( + "/platform/operations/deliver", who="operator", event_id=event.event_id + ) + token = self.confirm_token(body) + self.app.service.store.save_outbox(replace(event, delivery_attempts=1)) + response, _ = self.post( + "/platform/operations/deliver", + who="operator", + event_id=event.event_id, + confirm_token=token, + ) + self.assertEqual("409 Conflict", response["status"]) + self.assertEqual([], self.calls) + + def test_failure_releases_claim_and_retry_is_available(self): + event = self.seed() + + def fail(event): + raise RuntimeError("private-token-in-error") + + self.app.outbox_delivery = fail + self.post( + "/platform/operations/deliver", + who="operator", + event_id=event.event_id, + confirmed=True, + ) + failed = self.app.service.store.outbox_event(event.event_id) + self.assertIsNone(failed.claimed_by) + self.assertEqual("delivery_unavailable", failed.failure_reason) + _, body = invoke(self.app, "/platform/operations", cookie="ue_session=operator") + self.assertIn(b"Review delivery attempt", body) + self.assertNotIn(b"private-token", body) + self.app.outbox_delivery = lambda e: self.calls.append(e.event_id) + self.post( + "/platform/operations/deliver", + who="operator", + event_id=event.event_id, + confirmed=True, + ) + self.assertIsNotNone( + self.app.service.store.outbox_event(event.event_id).delivered_at + ) + + def test_crashed_claim_reconciles_under_event_guard(self): + event = self.seed() + self.app.service.store.save_outbox( + replace(event, claimed_by="crashed", claimed_at=utc_now()) + ) + self.post( + "/platform/operations/deliver", + who="operator", + event_id=event.event_id, + confirmed=True, + ) + self.assertEqual(["p05-one"], self.calls) + + def test_missing_mail_lane_never_marks_an_invitation_accepted(self): + event = self.seed() + adapter = HTTPOutboxDeliveryAdapter( + event_url="http://unused", event_bearer_token="fixture" + ) + with self.assertRaises(DeliveryError) as error: + adapter(event) + self.assertEqual("mail_unconfigured", error.exception.code) + + def test_status_is_checked_bounded_and_never_claims_receipt(self): + class Mail: + def mail_status(self): + return {"checks": {"database": "ok", "smtp_authentication": "failed"}} + + class OIDC: + backend_url = "http://fixture" + + with patch( + "user_engine.operations_status.sign_in_status", + return_value={ + "checks": [ + {"name": "factor_reader", "status": "failed"}, + {"name": "lldap", "status": "ok"}, + ] + }, + ): + html = check_services(OIDC(), Mail()) + self.assertIn("Factor-read credential", html) + self.assertIn("Unavailable", html) + self.assertIn("send no email", html) + self.assertIn("do not prove inbox receipt", html) + + def test_selected_mail_evidence_is_safe_and_read_only(self): + event = self.seed() + + class Mail: + def mail_delivery_status(self, event_id): + return {"state": "unknown", "private": "secret-provider-detail"} + + self.app.outbox_delivery = Mail() + _, body = invoke( + self.app, + "/platform/operations", + cookie="ue_session=operator", + query="event_id=" + event.event_id, + ) + self.assertIn(b"Automatic resending is blocked", body) + self.assertNotIn(b"secret-provider-detail", body) + self.assertEqual( + 0, self.app.service.store.outbox_event(event.event_id).delivery_attempts + ) diff --git a/workplans/USER-WP-0032-platform-service-operations.md b/workplans/USER-WP-0032-platform-service-operations.md new file mode 100644 index 0000000..a1815f7 --- /dev/null +++ b/workplans/USER-WP-0032-platform-service-operations.md @@ -0,0 +1,57 @@ +--- +id: USER-WP-0032 +type: workplan +title: "P05 checked service health and bounded delivery recovery" +domain: communication +repo: user-engine +status: active +owner: codex +topic_slug: communication +created: "2026-09-13" +updated: "2026-09-13" +--- + +Complete P05 under USER-WP-0030-T03. P06 remains separate. Do not send production +messages for acceptance; use the existing loopback SMTP/IMAP harness. Live SMTP +authentication diagnostics do not submit a message or claim inbox delivery. + +## Expose checked service state with clear recovery steps + +```task +id: USER-WP-0032-T01 +status: progress +priority: high +``` + +Operator-only, bounded checks for KeyCape dependencies/current factor-read +credential and mail database/SMTP authentication. Show checked time, impact, +unknown versus unavailable state, retry and assisted setup; no secrets or raw +provider errors. Health probes must not send mail. + +## Make selected delivery and retry safe and reviewable + +```task +id: USER-WP-0032-T02 +status: progress +priority: high +``` + +Fix failed worker claims; serialize event operations across connections; confirm +one selected delivery and reject stale/repeated completion. Missing email lane +must not mark invitations accepted. Preserve provider idempotency across lost +responses and restart; uncertain SMTP outcomes must not auto-resend. No automatic +processing of the production backlog during this task. + +## Verify failure and recovery through native and harness acceptance + +```task +id: USER-WP-0032-T03 +status: todo +priority: high +``` + +Installed-provider genuine JWT expiry, scope withdrawal/recovery and outage; +native renewal/projection; SMTP/IMAP received-message and retry tests with a +local mailbox; role/CSRF/stale/ambiguous-outcome browser coverage. Publish digest +pinned releases and verify live non-sending status. Close RPF-WP-0040 acceptance +using its specified isolated fixture rather than disrupting live authentication.