r""" eCW Charge Validation — results portal (2026-10-08; redesigned 2026-10-19, spec docs/superpowers/specs/2026-10-19-results-portal-redesign-design.md). Reads `Charge_Validation.dbo.CV_ChargeSummary` (written by the bot when RESULTS_DB=1). Four pages: the Dashboard (`/` — Bot run card, today's counts, Find a claim, Claim status), Claim status (`/days` — one card per day a run started), one day (`/day/YYYY-MM-DD` — every run of that day, a short table, each claim's fields in a side panel) and a claim's history (`/claim?claim_no=`). Old `/?run_at=` links go to that run's day. run_web.bat python -m uvicorn web.app:app --host 127.0.0.1 --port 8000 Bound to 127.0.0.1 only: the rows hold PHI (patient, account number), so the page is not reachable from the network. Read-only for the results; it can also start a VERIFY run of the bot and stop it (2026-10-14, `bot_runner.py`), each POST checked for this page's Origin and token. JSON for the later portal: /api/runs, /api/results. Hardening (review, 2026-10-12): only the Host names 127.0.0.1 / localhost are answered (a web page elsewhere whose name resolves to 127.0.0.1 — DNS rebinding — gets a 400, not the PHI); every response is `no-store` so no browser or proxy keeps a copy; a database error is reported by its kind (not reachable / login / table missing / permission), never with the driver's message. """ import re import sys import html import json import time import pathlib from collections import Counter from datetime import date, datetime, timedelta from functools import partial ROOT = pathlib.Path(__file__).resolve().parent.parent if str(ROOT / "src") not in sys.path: sys.path.insert(0, str(ROOT / "src")) from fastapi import Body, FastAPI, Form, Request # noqa: E402 from starlette.middleware.trustedhost import TrustedHostMiddleware # noqa: E402 from fastapi.encoders import jsonable_encoder # noqa: E402 from fastapi.responses import (FileResponse, HTMLResponse, # noqa: E402 JSONResponse, RedirectResponse) from fastapi.templating import Jinja2Templates # noqa: E402 import secrets # noqa: E402 from urllib.parse import urlencode, urlsplit # noqa: E402 from ecw_bot import settings as cfg # noqa: E402 from ecw_bot.core import results_db # noqa: E402 from ecw_bot.core.run_events import run_id_of # noqa: E402 from .repo import SqlResultsRepo # noqa: E402 from . import run_status # noqa: E402 from . import runs as run_view # noqa: E402 from .users import (Accounts, CAN_ADMIN, CAN_RUN, # noqa: E402 EcareSignIn, IDLE_MINUTES, LOCK_AFTER, LOCK_MINUTES, MANAGES, Sessions, SqlUserStore, UserError, is_ecare) from .bot_runner import (AlreadyRunning, BotRunner, # noqa: E402 NoRunner, NotRunning, StopFailed) CATEGORIES = ("Pass", "Fail", "Review") DB_DOWN = "Results database not reachable" ALLOWED_HOSTS = ("127.0.0.1", "localhost") # The Sign-in cookie (Stage 2, 2026-10-22). Every page but these needs a # signed-in Portal User. COOKIE = "ecw_portal" OPEN_PATHS = ("/signin", "/api/auth/login") SIGN_IN_ERRORS = { "bad": "Wrong username or password.", "locked": "Too many wrong passwords — this user is locked for 15 " "minutes.", "disabled": "This user is disabled — ask an Admin.", "down": "The Ecare sign-in service is not answering — try again " "shortly.", "no_access": "Your Ecare login is right, but you have no access to " "this Portal — ask an Admin to add you.", "many": "Too many sign-in tries from here — wait 15 minutes.", } SIGN_IN_STATUS = {"down": 503, "no_access": 403, "many": 429} # Sign-in tries, right or wrong, from one address per SIGN_IN_WINDOW # minutes (2026-10-27, with the Ecare sign-in). SIGN_IN_LIMIT = 20 SIGN_IN_WINDOW = 15 # The only pages allowed to Start / Stop a run: this one (2026-10-14). A # page on another site in the same browser could otherwise POST here. # The port must match the one in run_web.bat (8000). ALLOWED_ORIGINS = ("http://127.0.0.1:8000", "http://localhost:8000") # claim_no is NVARCHAR(20): a longer search can never match. CLAIM_MAX = 20 # SQLSTATE -> (title, what to check). Anything else is "not reachable". _DB_PROBLEMS = { "28000": ("Results database login failed", "The Windows login has no access to Charge_Validation yet — " "see the GRANT lines at the end of tools/sql/create_tables.sql."), "42S02": ("Results table is missing", "Run tools/sql/create_tables.sql in SSMS once."), "42S22": ("Results table needs an update", "Run tools/sql/create_tables.sql in SSMS once more — it adds " "the started_by column (2026-10-22)."), "42000": ("No permission to read the results table", "The Windows login needs db_datareader on Charge_Validation."), } _DB_DEFAULT = (DB_DOWN, "Check that SQL Server is running and that " "RESULTS_DB_CONN in .env points at the " "Charge_Validation database.") _SECURITY_HEADERS = { "Cache-Control": "no-store", "Pragma": "no-cache", "X-Frame-Options": "DENY", "X-Content-Type-Options": "nosniff", } TEMPLATES = Jinja2Templates( directory=str(pathlib.Path(__file__).resolve().parent / "templates")) def _money(value): return "" if value is None else f"{value:,.2f}" def _us_date(value): if isinstance(value, datetime): return value.strftime("%m/%d/%Y %H:%M:%S") if isinstance(value, date): return value.strftime("%m/%d/%Y") return "" if value is None else str(value) def _long_date(value): """Thu, 16 Oct 2026 — how a day reads on its card and page.""" return value.strftime("%a, %d %b %Y") if value else "" def _clock(value): """2:15 PM — a run's time of day.""" return value.strftime("%I:%M %p").lstrip("0") if value else "" TEMPLATES.env.filters["money"] = _money TEMPLATES.env.filters["usdate"] = _us_date TEMPLATES.env.filters["longdate"] = _long_date TEMPLATES.env.filters["clock"] = _clock def _duration(seconds): """690 -> 11m 30s; 3725 -> 1h 2m. Blank for none.""" if seconds is None: return "" seconds = int(seconds) h, rest = divmod(seconds, 3600) m, s = divmod(rest, 60) if h: return f"{h}h {m}m" return f"{m}m {s}s" if m else f"{s}s" TEMPLATES.env.filters["dur"] = _duration _MONEY_COLUMNS = {"charges", "pmts_adjs", "pmts_refunds", "adjustment", "withheld", "balance"} _DATE_COLUMNS = {"service_date", "received_date", "completed_on"} def _shown(col, value) -> str: """One Charge Summary field as the page shows it.""" if value is None: return "" if col in _MONEY_COLUMNS: return _money(value) if col in _DATE_COLUMNS: return _us_date(value) return str(value) class BadFilter(Exception): pass class Forbidden(Exception): pass class DbUnavailable(Exception): def __init__(self, title=DB_DOWN, hint=_DB_DEFAULT[1]): super().__init__(title) self.title, self.hint = title, hint def _filters(run_at, category, claim_no): """Validated (run_at datetime|None, category|None, claim digits|None).""" when = None if run_at: try: when = datetime.fromisoformat(run_at) except ValueError: raise BadFilter(f"run_at is not a date/time: {run_at!r}") # run_at is saved as the PC's local time, with no zone. when = when.replace(tzinfo=None) category = (category or "").strip() or None if category and category not in CATEGORIES: raise BadFilter(f"category must be one of {', '.join(CATEGORIES)}") claim = re.sub(r"[^0-9A-Za-z]", "", claim_no or "")[:CLAIM_MAX] or None return when, category, claim def _db(fn, *args, **kwargs): try: return fn(*args, **kwargs) except UserError: raise # the user's to fix, not the database's except Exception as exc: state = exc.args[0] if exc.args and isinstance(exc.args[0], str) \ else "" title, hint = _DB_PROBLEMS.get(state, _DB_DEFAULT) print(f"[WEB] {title}: {results_db.safe_error(exc)}") raise DbUnavailable(title, hint) from exc def _details(row: dict) -> dict: try: return json.loads(row.get("details_json") or "{}") except ValueError: return {"unreadable": row.get("details_json")} def _day(text) -> date: try: return date.fromisoformat(text) except (TypeError, ValueError): raise BadFilter(f"not a day (YYYY-MM-DD): {text!r}") def _for_panel(row: dict) -> dict: """A results row with what the page's side panel shows: every Charge Summary field as text, the run's time and mode, the details JSON.""" details = _details(row) row["fields"] = ([[h, _shown(c, row.get(c))] for h, c in results_db.COLUMNS] + [["Run", _us_date(row.get("run_at"))], ["Mode", row.get("run_mode") or ""], ["Started By", row.get("started_by") or ""]]) row["details_pretty"] = (json.dumps(details, indent=2, ensure_ascii=False) if details else "") return row def _counts(rows) -> dict: out = {"claims": len(rows), "runs": len({r.get("run_at") for r in rows})} for cat in CATEGORIES: out[cat.lower() + "_count"] = sum(r.get("category") == cat for r in rows) return out def _reasons(rows, top=6) -> list: """A day's Fail and Review comments, grouped: `[{category, comment, count}]`, most frequent first, Fail before Review on a tie.""" counted = Counter((r.get("category"), (r.get("comments") or "").strip()) for r in rows if r.get("category") in ("Fail", "Review")) ordered = sorted(counted.items(), key=lambda kv: (-kv[1], kv[0][0] != "Fail", kv[0][1])) return [{"category": cat, "comment": text or "(no comment)", "count": n} for (cat, text), n in ordered[:top]] def portal_settings(urls: str, server: bool) -> dict: """create_app's host / origin / cookie / Start settings from PORTAL_URLS (comma separated, as people type them in the browser) and PORTAL_SERVER. No URLs: this PC only. A server Portal never starts the bot; an all-https Portal sends the cookie only over https.""" parts = [urlsplit(u.strip()) for u in (urls or "").split(",") if u.strip()] if not parts: return {"allowed_hosts": list(ALLOWED_HOSTS), "allowed_origins": list(ALLOWED_ORIGINS), "secure_cookie": False, "can_start": not server} return {"allowed_hosts": [u.hostname for u in parts], "allowed_origins": [f"{u.scheme}://{u.netloc}" for u in parts], "secure_cookie": all(u.scheme == "https" for u in parts), "can_start": not server} def create_app(repo, allowed_hosts=ALLOWED_HOSTS, runner=None, allowed_origins=ALLOWED_ORIGINS, logs_dir=None, screenshots_root=None, users_store=None, sessions=None, can_start=True, secure_cookie=False, ecare=None) -> FastAPI: app = FastAPI(title="eCW Charge Validation — Results", docs_url=None, redoc_url=None, openapi_url=None) if runner is None: runner = BotRunner() if can_start else NoRunner() logs_dir = logs_dir if logs_dir is not None else cfg.LOGS screenshots_root = (screenshots_root if screenshots_root is not None else cfg.ROOT / "Screenshots") # Made per server start and put in the page: a page on another site # cannot read it, so it cannot send it. app.state.run_token = secrets.token_urlsafe(32) store = (users_store if users_store is not None else SqlUserStore(cfg.RESULTS_DB_CONN)) accounts = Accounts(store, ecare=ecare) sessions = sessions if sessions is not None else Sessions() app.state.sessions = sessions sign_in_tries = {} # client address -> recent try times def _too_many(request: Request) -> bool: """Counts this try; True past SIGN_IN_LIMIT in SIGN_IN_WINDOW.""" who = request.client.host if request.client else "?" now = time.monotonic() recent = [t for t in sign_in_tries.get(who, ()) if now - t < SIGN_IN_WINDOW * 60] recent.append(now) sign_in_tries[who] = recent return len(recent) > SIGN_IN_LIMIT def _start_session(response, user) -> None: response.set_cookie(COOKIE, sessions.new(user), httponly=True, samesite="strict", path="/", secure=secure_cookie) def _check_token(request: Request, sent: str) -> None: origin = request.headers.get("origin") if not origin: ref = urlsplit(request.headers.get("referer") or "") origin = f"{ref.scheme}://{ref.netloc}" if ref.netloc else "" if origin not in allowed_origins: raise Forbidden() try: same = secrets.compare_digest(sent or "", app.state.run_token) except TypeError: # non-ASCII text same = False if not same: raise Forbidden() def _check_writer(request: Request) -> None: """A script's POST: the token in its X-Run-Token header.""" _check_token(request, request.headers.get("x-run-token")) def _need(request: Request, roles) -> dict: user = request.state.user if user["role"] not in roles: raise Forbidden() return user # Before _no_store, so its redirects get those headers too. @app.middleware("http") async def _signed_in(request: Request, call_next): path = request.url.path if path in OPEN_PATHS: request.state.user = None return await call_next(request) user, timed_out = sessions.get(request.cookies.get(COOKIE)) if timed_out: try: store.record(timed_out, "timeout") except Exception: pass if user is None: if path.startswith("/api"): return JSONResponse({"detail": "Sign in first"}, status_code=401) back = path + (f"?{request.url.query}" if request.url.query else "") return RedirectResponse("/signin?" + urlencode({"next": back}), status_code=303) if user["must_change"] and path not in ("/account/password", "/signout"): if path.startswith("/api"): return JSONResponse({"detail": "Change your password first"}, status_code=403) return RedirectResponse("/account/password", status_code=303) request.state.user = user return await call_next(request) @app.middleware("http") async def _no_store(request: Request, call_next): response = await call_next(request) response.headers.update(_SECURITY_HEADERS) return response # Added last, so it runs first: a foreign Host gets a bare 400 before # any route (or the database) is reached. app.add_middleware(TrustedHostMiddleware, allowed_hosts=list(allowed_hosts)) @app.exception_handler(BadFilter) def _bad_filter(request: Request, exc: BadFilter): if request.url.path.startswith("/api"): return JSONResponse({"detail": str(exc)}, status_code=400) # The message quotes what was typed in the URL: escaped, never # echoed raw. return HTMLResponse(f"

Bad filter

{html.escape(str(exc))}" f"

", status_code=400) @app.exception_handler(Forbidden) def _forbidden(request: Request, exc: Forbidden): if request.url.path.startswith("/api"): return JSONResponse({"detail": "Not allowed from here"}, status_code=403) return HTMLResponse("

Not allowed

Your role cannot do " "this, or the request did not come from this " "page.

", status_code=403) @app.exception_handler(DbUnavailable) def _db_down(request: Request, exc: DbUnavailable): if request.url.path.startswith("/api"): return JSONResponse({"detail": exc.title}, status_code=503) return HTMLResponse(f"

{html.escape(exc.title)}

" f"

{html.escape(exc.hint)}

", status_code=503) def _page(request, name, crumbs, **context): """Every page: its template, the mode pill, the run token, and the breadcrumb (`[label, link]`, the last one without a link).""" user = getattr(request.state, "user", None) return TEMPLATES.TemplateResponse(request, name, { "mode": "LIVE-WRITE" if cfg.SAVE_CORRECTIONS else "VERIFY", "run_token": app.state.run_token, "crumbs": crumbs, "categories": CATEGORIES, "user": user, "can_run": bool(can_start and user and user["role"] in CAN_RUN), "is_admin": bool(user and user["role"] in CAN_ADMIN), **context}) @app.get("/", response_class=HTMLResponse) def dashboard(request: Request, run_at: str = None, category: str = None, claim_no: str = None): when, cat, claim = _filters(run_at, category, claim_no) if when is not None: # An old one-run link: that run's day. query = urlencode({k: v for k, v in (("category", cat), ("claim_no", claim)) if v}) return RedirectResponse( f"/day/{when.date().isoformat()}" + (f"?{query}" if query else ""), status_code=303) days = _db(repo.days) today = date.today() counts = next((d for d in days if d["day"] == today), {"day": today, "runs": 0, **_counts([])}) week = sorted((d for d in days if today - timedelta(days=6) <= d["day"] <= today), key=lambda d: d["day"]) # Only a day with claims costs the second query. reasons = (_reasons(_db(repo.day_results, today)) if counts["claims"] else []) return _page(request, "dashboard.html", [["Dashboard", None]], today=counts, has_runs=bool(days), alert=_last_run_alert(), week=week, reasons=reasons) @app.get("/days", response_class=HTMLResponse) def days_page(request: Request): return _page(request, "days.html", [["Dashboard", "/"], ["Claim status", None]], days=_db(repo.days)) @app.get("/day/{day}", response_class=HTMLResponse) def day_page(request: Request, day: str, category: str = None, claim_no: str = None, claim: str = None): when = _day(day) _, cat, search = _filters(None, category, claim_no) _, _, opened = _filters(None, None, claim) rows = _db(repo.day_results, when) shown = [_for_panel(r) for r in rows if (not cat or r.get("category") == cat) and (not search or search in str(r.get("claim_no") or ""))] return _page(request, "day.html", [["Dashboard", "/"], ["Claim status", "/days"], [_long_date(when), None]], day=when, counts=_counts(rows), rows=shown, category=cat or "", claim_no=search or "", opened=opened or "") @app.get("/claim", response_class=HTMLResponse) def claim_page(request: Request, claim_no: str = None): _, _, claim = _filters(None, None, claim_no) rows = ([_for_panel(r) for r in _db(repo.claim_history, claim)] if claim else []) return _page(request, "claim.html", [["Dashboard", "/"], ["Find a claim", None]], rows=rows, claim_no=claim or "") def _last_run_alert(): """The latest Run, if it ended badly — the Dashboard's red banner (EEWA-33). A Run Error, or a Run that ended without a record (killed); never a Stop, a finished Run or one still going.""" last = run_view.latest_run(logs_dir, _live_run_id()) if last and last["status"] in ("Run Error", "Ended without a record"): return last return None def _live_run_id(): """The Run whose process is alive now, if any (Runs page).""" try: st = runner.status() except Exception: return None if st.get("running") and st.get("log_path"): return run_id_of(st["log_path"]) return None @app.get("/runs", response_class=HTMLResponse) def runs_page(request: Request, day_from: str = None, day_to: str = None, status: str = None): for value in (day_from, day_to): if value: try: date.fromisoformat(value) except ValueError: raise BadFilter(f"Not a date: {value}") if status and status not in run_view.STATUSES: raise BadFilter(f"Not a result: {status}") rows = run_view.list_runs(logs_dir, _live_run_id(), day_from=day_from or None, day_to=day_to or None, status=status or None) return _page(request, "runs.html", [["Dashboard", "/"], ["Runs", None]], runs=rows, statuses=run_view.STATUSES, day_from=day_from or "", day_to=day_to or "", status=status or "") @app.get("/runs/{run_id}", response_class=HTMLResponse) def run_page(request: Request, run_id: str): run = run_view.load_run(logs_dir, run_id, _live_run_id()) if run is None: return HTMLResponse("

No such run

", status_code=404) return _page(request, "run.html", [["Dashboard", "/"], ["Runs", "/runs"], [_us_date(run["started"]), None]], run=run) @app.get("/runs/{run_id}/shot/{n}") def run_shot(run_id: str, n: int): path = run_view.screenshot_path(logs_dir, run_id, n, screenshots_root) if path is None: return HTMLResponse("

No such screenshot

", status_code=404) return FileResponse(path) @app.get("/api/runs") def api_runs(): return jsonable_encoder(_db(repo.runs)) @app.get("/api/results") def api_results(run_at: str = None, category: str = None, claim_no: str = None): when, cat, claim = _filters(run_at, category, claim_no) if when is None: runs = _db(repo.runs) if not runs: return {"run_at": None, "rows": []} when = runs[0]["run_at"] rows = [] for row in _db(repo.results, when, cat, claim): out = {k: v for k, v in row.items() if k != "details_json"} out["details"] = _details(row) rows.append(out) return jsonable_encoder({"run_at": when, "rows": rows}) @app.post("/api/bot/start") def bot_start(request: Request): _check_writer(request) user = _need(request, CAN_RUN if can_start else ()) try: return run_status.summarize( runner.start(started_by=user["username"])) except AlreadyRunning as exc: return JSONResponse({"detail": str(exc)}, status_code=409) @app.post("/api/bot/stop") def bot_stop(request: Request): _check_writer(request) user = _need(request, CAN_RUN if can_start else ()) try: return run_status.summarize( runner.stop(stopped_by=user["username"])) except NotRunning as exc: return JSONResponse({"detail": str(exc)}, status_code=409) except StopFailed as exc: return JSONResponse({"detail": str(exc)}, status_code=500) @app.get("/api/bot/status") def bot_status(): return run_status.summarize(runner.status()) # ── Sign-in (Stage 2, EEWA-32) ─────────────────────────────────────── def _signin_page(request, error="", next_url="/", status_code=200): page = _page(request, "signin.html", [["Sign in", None]], error=error, next_url=_safe_next(next_url), no_users=not _db(store.list), ecare_on=accounts.ecare is not None, idle_minutes=IDLE_MINUTES, lock_after=LOCK_AFTER, lock_minutes=LOCK_MINUTES) page.status_code = status_code return page @app.get("/signin", response_class=HTMLResponse) def signin_page(request: Request, next: str = "/"): user, _ = sessions.get(request.cookies.get(COOKIE)) if user: return RedirectResponse(_safe_next(next), status_code=303) return _signin_page(request, next_url=next) @app.post("/signin", response_class=HTMLResponse) def signin(request: Request, username: str = Form(""), password: str = Form(""), token: str = Form(""), next: str = Form("/")): _check_token(request, token) why = "many" if _too_many(request) else None user = None if why is None: user, why = _db(accounts.sign_in, username, password) if user is None: return _signin_page(request, SIGN_IN_ERRORS[why], next, SIGN_IN_STATUS.get(why, 401)) # A new token every Sign-in: an old cookie never comes back to life. sessions.drop(request.cookies.get(COOKIE)) target = ("/account/password" if user["must_change"] else _safe_next(next)) response = RedirectResponse(target, status_code=303) _start_session(response, user) return response # The same Sign-in as JSON (2026-10-27): this page's token and Origin, # like every other write; the Ecare token never reaches the browser. def _who(user) -> dict: return {"username": user["username"], "display_name": user.get("display_name") or user["username"], "role": user["role"], "must_change": bool(user.get("must_change"))} @app.post("/api/auth/login") def auth_login(request: Request, payload: dict = Body(None)): _check_writer(request) if _too_many(request): return JSONResponse({"detail": SIGN_IN_ERRORS["many"]}, status_code=429) payload = payload if isinstance(payload, dict) else {} username = str(payload.get("username") or "").strip() password = str(payload.get("password") or "") if not username or not password.strip(): return JSONResponse({"detail": "Username and password are both " "needed"}, status_code=400) user, why = _db(accounts.sign_in, username, password) if user is None: return JSONResponse({"detail": SIGN_IN_ERRORS[why]}, status_code=SIGN_IN_STATUS.get(why, 401)) sessions.drop(request.cookies.get(COOKIE)) response = JSONResponse(_who(user)) _start_session(response, user) return response @app.get("/api/auth/me") def auth_me(request: Request): return _who(request.state.user) @app.post("/api/auth/logout") def auth_logout(request: Request): _check_writer(request) _db(store.record, request.state.user["username"], "sign_out") sessions.drop(request.cookies.get(COOKIE)) response = JSONResponse({"signed_out": True}) response.delete_cookie(COOKIE, path="/") return response @app.post("/signout") def signout(request: Request, token: str = Form("")): _check_token(request, token) _db(store.record, request.state.user["username"], "sign_out") sessions.drop(request.cookies.get(COOKIE)) response = RedirectResponse("/signin", status_code=303) response.delete_cookie(COOKIE, path="/") return response def _password_page(request, error="", status_code=200): page = _page(request, "password.html", [["Dashboard", "/"], ["Change password", None]], error=error) page.status_code = status_code return page @app.get("/account/password", response_class=HTMLResponse) def password_page(request: Request): return _password_page(request) @app.post("/account/password", response_class=HTMLResponse) def password_change(request: Request, current: str = Form(""), new: str = Form(""), confirm: str = Form(""), token: str = Form("")): _check_token(request, token) try: if new != confirm: raise UserError("The two new passwords are not the same") updated = _db(accounts.change_password, request.state.user["username"], current, new) except UserError as exc: return _password_page(request, str(exc), 400) sessions.refresh_user(updated) return RedirectResponse("/", status_code=303) # ── Portal Users (Super Admin, Admin) ──────────────────────────────── def _users_page(request, error="", message="", status_code=200): page = _page(request, "users.html", [["Dashboard", "/"], ["Portal Users", None]], users=[{**u, "ecare": is_ecare(u)} for u in _db(store.list)], roles=MANAGES.get(request.state.user["role"], ()), error=error, message=message) page.status_code = status_code return page @app.get("/admin/users", response_class=HTMLResponse) def users_page(request: Request, msg: str = ""): _need(request, CAN_ADMIN) return _users_page(request, message=msg[:200]) def _admin_change(request, token, done, fn, *args): """One Admin change: checked, made, then back to the list with `done` said — or the list again with the problem.""" _check_token(request, token) me = _need(request, CAN_ADMIN)["username"] try: _db(fn, *args, by=me) except UserError as exc: return None, _users_page(request, error=str(exc), status_code=400) return True, RedirectResponse( "/admin/users?" + urlencode({"msg": done}), status_code=303) @app.post("/admin/users", response_class=HTMLResponse) def users_create(request: Request, username: str = Form(""), display_name: str = Form(""), role: str = Form(""), password: str = Form(""), signin: str = Form("ecare"), token: str = Form("")): name = username.strip().lower() if signin != "portal": return _admin_change( request, token, f"{name} added — they sign in with their " f"Ecare office login", partial(accounts.create_user, ecare=True), username, display_name, role, None)[1] return _admin_change( request, token, f"{name} added — they choose their own " f"password at their first Sign-in", accounts.create_user, username, display_name, role, password)[1] @app.post("/admin/users/{name}/reset", response_class=HTMLResponse) def users_reset(request: Request, name: str, password: str = Form(""), token: str = Form("")): ok, page = _admin_change( request, token, f"{name}: password reset — they choose a new " f"one at their next Sign-in", accounts.reset_password, name, password) if ok: sessions.drop_user(name) return page @app.post("/admin/users/{name}/role", response_class=HTMLResponse) def users_role(request: Request, name: str, role: str = Form(""), token: str = Form("")): ok, page = _admin_change(request, token, f"{name} is now {role}", accounts.set_role, name, role) if ok: updated = _db(store.get, name) if updated: sessions.refresh_user(updated) return page @app.post("/admin/users/{name}/disable", response_class=HTMLResponse) def users_disable(request: Request, name: str, token: str = Form("")): ok, page = _admin_change(request, token, f"{name} disabled", accounts.set_disabled, name, True) if ok: sessions.drop_user(name) return page @app.post("/admin/users/{name}/enable", response_class=HTMLResponse) def users_enable(request: Request, name: str, token: str = Form("")): return _admin_change(request, token, f"{name} enabled", accounts.set_disabled, name, False)[1] @app.get("/admin/signins", response_class=HTMLResponse) def signins_page(request: Request): _need(request, CAN_ADMIN) return _page(request, "signins.html", [["Dashboard", "/"], ["Sign-in history", None]], events=_db(store.history)) return app def _safe_next(target) -> str: """Only a path on this site: never `//elsewhere`, `/\\elsewhere` or a full URL.""" target = target or "/" if not target.startswith("/") or target[1:2] in ("/", "\\"): return "/" return target app = create_app(SqlResultsRepo(cfg.RESULTS_DB_CONN), users_store=SqlUserStore(cfg.RESULTS_DB_CONN), ecare=(EcareSignIn(cfg.AUTH_TOKEN_API_URL) if cfg.AUTH_TOKEN_API_URL else None), **portal_settings(cfg.PORTAL_URLS, cfg.PORTAL_SERVER))