diff --git a/egomimic/ingest/accounts.py b/egomimic/ingest/accounts.py new file mode 100644 index 000000000..f4e052c87 --- /dev/null +++ b/egomimic/ingest/accounts.py @@ -0,0 +1,155 @@ +"""Dashboard accounts, kept in the bucket (pipeline/accounts.json). + + python -m egomimic.ingest.accounts add --vendor | --staff + python -m egomimic.ingest.accounts password + python -m egomimic.ingest.accounts remove + python -m egomimic.ingest.accounts list + +A vendor account sees only its vendor's deliveries; a staff account sees every vendor +and (later) approves. We create the accounts: there is no sign-up. Passwords are stored +as salted scrypt hashes, never in the clear. Store flags as for egomimic.ingest.watch. +""" + +from __future__ import annotations + +import argparse +import getpass +import hashlib +import hmac +import json +import os +import sys + +from egomimic.ingest.store import VENDOR, LocalStore, R2Store, Store, now + +ROLES = ("vendor", "staff") +MIN_PASSWORD = 12 +# scrypt cost: ~50 ms and 16 MiB per check, so a stolen accounts file is slow to crack. +SCRYPT = {"n": 2**14, "r": 8, "p": 1} + + +def hash_password(password: str, salt: bytes | None = None) -> dict: + salt = salt or os.urandom(16) + digest = hashlib.scrypt(password.encode(), salt=salt, dklen=32, **SCRYPT) + return {"scrypt": {**SCRYPT, "salt": salt.hex(), "hash": digest.hex()}} + + +def check_password(account: dict | None, password: str) -> bool: + """Constant-time; an unknown user costs the same as a wrong password.""" + rec = (account or {}).get("password", {}).get("scrypt") + if rec is None: + hash_password(password) # same work as a real check + return False + params = {k: rec[k] for k in ("n", "r", "p")} + digest = hashlib.scrypt( + password.encode(), salt=bytes.fromhex(rec["salt"]), dklen=32, **params + ) + return hmac.compare_digest(digest.hex(), rec["hash"]) + + +def add( + store: Store, user: str, password: str, *, role: str, vendor: str | None = None +) -> dict: + if not (VENDOR.fullmatch(user) and len(user) <= 64): # also the reviewer name + raise ValueError(f"{user!r}: lowercase letters and digits, joined by _ or -") + if role not in ROLES: + raise ValueError(f"role must be one of {ROLES}") + if role == "vendor" and not (vendor and VENDOR.fullmatch(vendor)): + raise ValueError("a vendor account needs a vendor slug") + if role == "staff" and vendor: + raise ValueError("a staff account has no vendor") + if len(password) < MIN_PASSWORD: + raise ValueError(f"passwords need at least {MIN_PASSWORD} characters") + accounts = store.get_accounts() + if user in accounts: + raise ValueError(f"{user} already exists") + accounts[user] = { + "role": role, + "vendor": vendor if role == "vendor" else None, + "password": hash_password(password), + "created": now(), + } + store.put_accounts(accounts) + return accounts[user] + + +def set_password(store: Store, user: str, password: str) -> None: + if len(password) < MIN_PASSWORD: + raise ValueError(f"passwords need at least {MIN_PASSWORD} characters") + accounts = store.get_accounts() + if user not in accounts: + raise KeyError(user) + accounts[user]["password"] = hash_password(password) + accounts[user]["password_set"] = now() + store.put_accounts(accounts) + + +def remove(store: Store, user: str) -> None: + accounts = store.get_accounts() + if accounts.pop(user, None) is None: + raise KeyError(user) + store.put_accounts(accounts) + + +def store_from_args(args) -> Store: + if args.local_store: + return LocalStore(args.local_store) + return R2Store(args.bucket, root=args.root) + + +def add_store_args(ap: argparse.ArgumentParser) -> None: + src = ap.add_mutually_exclusive_group() + src.add_argument("--bucket", default="rldb", help="R2 bucket (default rldb)") + src.add_argument("--local-store", help="a directory instead of R2") + ap.add_argument("--root", default="", help="R2 key prefix, e.g. _test/run1/") + + +def _new_password(user: str) -> str: + first = getpass.getpass(f"password for {user}: ") + if first != getpass.getpass("again: "): + raise SystemExit("passwords differ") + return first + + +def main(argv: list[str] | None = None) -> int: + ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0]) + add_store_args(ap) + sub = ap.add_subparsers(dest="cmd", required=True) + a = sub.add_parser("add") + a.add_argument("user") + who = a.add_mutually_exclusive_group(required=True) + who.add_argument("--vendor") + who.add_argument("--staff", action="store_true") + sub.add_parser("password").add_argument("user") + sub.add_parser("remove").add_argument("user") + sub.add_parser("list") + args = ap.parse_args(argv) + store = store_from_args(args) + try: + if args.cmd == "add": + role = "staff" if args.staff else "vendor" + add( + store, + args.user, + _new_password(args.user), + role=role, + vendor=args.vendor, + ) + elif args.cmd == "password": + set_password(store, args.user, _new_password(args.user)) + elif args.cmd == "remove": + remove(store, args.user) + else: + listing = { + u: {k: a.get(k) for k in ("role", "vendor", "created")} + for u, a in sorted(store.get_accounts().items()) + } + print(json.dumps(listing, indent=2)) + except (ValueError, KeyError) as e: + print(f"error: {e}", file=sys.stderr) + return 1 + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/egomimic/ingest/dashboard/__init__.py b/egomimic/ingest/dashboard/__init__.py new file mode 100644 index 000000000..8a08aa124 --- /dev/null +++ b/egomimic/ingest/dashboard/__init__.py @@ -0,0 +1 @@ +"""The ingestion dashboard (Flask): see app.py.""" diff --git a/egomimic/ingest/dashboard/__main__.py b/egomimic/ingest/dashboard/__main__.py new file mode 100644 index 000000000..808a7f3f6 --- /dev/null +++ b/egomimic/ingest/dashboard/__main__.py @@ -0,0 +1,5 @@ +import sys + +from egomimic.ingest.dashboard.app import main + +sys.exit(main()) diff --git a/egomimic/ingest/dashboard/app.py b/egomimic/ingest/dashboard/app.py new file mode 100644 index 000000000..12fb06ae2 --- /dev/null +++ b/egomimic/ingest/dashboard/app.py @@ -0,0 +1,349 @@ +"""The ingestion dashboard: vendors follow their own deliveries, staff see every vendor. + + python -m egomimic.ingest.dashboard [--local-store DIR | --bucket rldb --root PREFIX] + +Configuration comes from the environment, so it runs the same locally and on AWS: +EGOVERSE_DASHBOARD_SECRET (required: signs session cookies), R2_* credentials (read +from ~/.egoverse_env when present), EGOVERSE_DASHBOARD_INSECURE=1 to allow cookies +over plain http when running on localhost, and EGOVERSE_DASHBOARD_PROXIES= when n +trusted proxies (the AWS load balancer) forward requests, so login throttling sees the +client's address rather than the proxy's. + +The dashboard only reads and writes the ingest buckets (accounts, statuses, uploads into +each vendor's bucket, decisions); the watcher on the ingest box does the work +(egomimic/ingest/aws/README.md). +""" + +from __future__ import annotations + +import hmac +import os +import secrets +import time +from datetime import timedelta +from functools import wraps +from pathlib import Path + +from flask import ( + Flask, + abort, + flash, + g, + jsonify, + redirect, + render_template, + request, + session, + url_for, +) +from jinja2 import DictLoader +from werkzeug.exceptions import HTTPException + +from egomimic.ingest import accounts, formats +from egomimic.ingest.dashboard import review, templates, uploads, views +from egomimic.ingest.dashboard.throttle import Throttle +from egomimic.ingest.store import Ref, Store + +ACCOUNTS_TTL = 10.0 # seconds an accounts read is reused across requests + + +def create_app( + store: Store, + secret_key: str, + *, + secure_cookies: bool = True, + proxies: int = 0, + throttle: Throttle | None = None, +) -> Flask: + """`proxies`: how many trusted proxies (e.g. the AWS load balancer) sit in front, + so the client address is taken from their X-Forwarded-For; 0 trusts none.""" + if not secret_key or len(secret_key) < 32: + raise ValueError("the session secret needs at least 32 characters") + app = Flask(__name__) + app.jinja_loader = DictLoader(templates.TEMPLATES) # *.html: autoescaped + app.jinja_env.filters.update(templates.FILTERS) + app.config.update( + SECRET_KEY=secret_key, + SESSION_COOKIE_HTTPONLY=True, + SESSION_COOKIE_SAMESITE="Lax", + SESSION_COOKIE_SECURE=secure_cookies, + PERMANENT_SESSION_LIFETIME=timedelta(hours=12), + MAX_CONTENT_LENGTH=64 * 1024, # forms only; files go straight to the bucket + ) + if proxies: + from werkzeug.middleware.proxy_fix import ProxyFix + + app.wsgi_app = ProxyFix(app.wsgi_app, x_for=proxies, x_proto=proxies) + app.extensions["ingest_store"] = store + throttle = throttle or Throttle() + # The upload page sends parts straight to the bucket's presigned links. + connect_src = " ".join(["'self'", *store.upload_origins]) + cache: dict = {"at": -1e9, "accounts": {}} + + def load_accounts() -> dict: + if time.monotonic() - cache["at"] > ACCOUNTS_TTL: + cache.update(at=time.monotonic(), accounts=store.get_accounts()) + return cache["accounts"] + + def password_mark(account: dict) -> str: + """Changes with the password, so a reset ends the account's sessions.""" + return account["password"]["scrypt"]["hash"][:16] + + @app.before_request + def who(): + g.user = g.account = None + user = session.get("user") + if user: + account = load_accounts().get(user) + if account and hmac.compare_digest( + session.get("mark", ""), password_mark(account) + ): + g.user, g.account = user, account + else: # removed, or the password changed since login + session.clear() + if "csrf" not in session: + session["csrf"] = secrets.token_urlsafe(32) + if request.method in ("POST", "PUT", "DELETE") and request.path != "/api/login": + sent = request.headers.get("X-CSRF-Token") or request.form.get("csrf", "") + if not hmac.compare_digest(sent, session["csrf"]): + abort(400, "form expired; reload the page") + + @app.after_request + def headers(resp): + if getattr(g, "retry_after", None): + resp.headers.setdefault("Retry-After", str(g.retry_after)) + resp.headers["Content-Security-Policy"] = ( + "default-src 'self'; frame-ancestors 'none'; form-action 'self'; " + f"connect-src {connect_src}" + ) + resp.headers["X-Content-Type-Options"] = "nosniff" + resp.headers["Referrer-Policy"] = "same-origin" + resp.headers["Cache-Control"] = "no-store" + return resp + + @app.context_processor + def common(): + return { + "user": g.get("user"), + "account": g.get("account"), + "csrf": session.get("csrf", ""), + "STATES": views.STATES, + "staff": is_staff(), + } + + def is_staff() -> bool: + return bool(g.get("account")) and g.account["role"] == "staff" + + def login_required(f): + @wraps(f) + def inner(*a, **kw): + if g.user is None: + return redirect(url_for("login", next=request.path)) + return f(*a, **kw) + + return inner + + def visible(vendor: str) -> bool: + return is_staff() or g.account.get("vendor") == vendor + + @app.get("/login") + def login(): + return render_template("login.html", error="") + + def log_in(user: str, password: str) -> tuple[bool, str, int]: + """Check and start a session: (ok, error, http status). The form and the CLI's + /api/login share it, throttling included.""" + user = user.strip().lower() + ip = request.remote_addr or "?" + wait = throttle.retry_after(ip, user) # before scrypt: cheap to refuse + if wait: + g.retry_after = wait + minutes = (wait + 59) // 60 + return False, f"Too many attempts. Try again in {minutes} minute(s).", 429 + account = load_accounts().get(user) + if not accounts.check_password(account, password): + throttle.failed(ip, user) + return False, "Wrong user or password.", 401 + throttle.succeeded(ip, user) + session.clear() # a fresh session on login, with a fresh CSRF token + session.permanent = True + session.update( + user=user, mark=password_mark(account), csrf=secrets.token_urlsafe(32) + ) + return True, "", 200 + + app.extensions["ingest_login"] = log_in + + @app.post("/login") + def login_post(): + ok, error, status = log_in( + request.form.get("user", ""), request.form.get("password", "") + ) + if not ok: + extra = {"Retry-After": str(g.retry_after)} if status == 429 else {} + return render_template("login.html", error=error), status, extra + target = request.args.get("next", "") + if not target.startswith("/") or target.startswith("//"): + target = url_for("index") # only our own pages + return redirect(target) + + @app.post("/logout") + def logout(): + session.clear() + return redirect(url_for("login")) + + @app.get("/style.css") + def style(): + return templates.CSS, 200, {"Content-Type": "text/css; charset=utf-8"} + + @app.get("/review.js") + def review_js(): + return templates.REVIEW_JS, 200, {"Content-Type": "text/javascript"} + + @app.get("/upload.js") + def upload_js(): + return templates.UPLOAD_JS, 200, {"Content-Type": "text/javascript"} + + @app.get("/upload.py") + def upload_cli(): + """The command-line uploader, one standard-library file vendors can run as is.""" + from egomimic.ingest import upload + + source = Path(upload.__file__).read_text() + return ( + source, + 200, + { + "Content-Type": "text/x-python; charset=utf-8", + "Content-Disposition": "attachment; filename=egoverse_upload.py", + }, + ) + + @app.get("/new") + @login_required + def new(): + if is_staff(): # vendors upload; staff review + flash("Uploads are made from vendor accounts.", "error") + return redirect(url_for("index")) + delivery_id = request.args.get("delivery", "") + _, limit = uploads.accepted() + return render_template( + "new.html", + delivery_id=delivery_id, + formats=formats.formats(), + limit_gib=limit // 1024**3, + ) + + @app.get("/") + @login_required + def index(): + if is_staff(): + vendor = request.args.get("vendor") or None + if vendor is not None and not _slug(vendor): + abort(404) + else: + vendor = g.account["vendor"] + listing = views.rows(store, vendor) + summary = [ + ("Total deliveries", len(listing)), + ("Awaiting review", sum(r.state == "ready" for r in listing)), + ("Committed", sum(r.state == "committed" for r in listing)), + ( + "Needs attention", + sum(r.state in {"failed", "rejected", "unsupported"} for r in listing), + ), + ] + state = request.args.get("state") + if state: + listing = [r for r in listing if r.state == state] + vendors = ( + sorted({r.ref.vendor for r in views.rows(store)}) if is_staff() else [] + ) + return render_template( + "index.html", + rows=listing, + vendor=vendor, + state=state, + vendors=vendors, + summary=summary, + ) + + @app.get("/d//") + @login_required + def delivery(vendor, delivery_id): + try: + ref = Ref(vendor, delivery_id) + except ValueError: + abort(404) + if not visible(vendor): + abort(404) # the same answer as a delivery that does not exist + r = views.row(store, ref) + if r is None: + abort(404) + st = r.status + # A decision on the current stage the watcher has not acted on yet, else the one + # the watcher recorded with its outcome. + decision = r.decision or st.get("decision") or {} + return render_template( + "delivery.html", + r=r, + error=st.get("error", "") if is_staff() else views.vendor_error(st), + decision=decision if is_staff() else views.vendor_decision(decision), + decisions=store.decisions(ref) if is_staff() else [], + ) + + @app.errorhandler(HTTPException) + def error_page(e: HTTPException): + """Errors in the dashboard's own layout; the API keeps its JSON answers.""" + if e.response is not None: # a full response was given (the API's JSON errors) + return e.response + if request.path.startswith("/api/"): + return jsonify({"error": e.description or e.name}), e.code + messages = { + 404: "There is no such page, or it is not one your account can see.", + 400: e.description or "That request was not understood.", + } + text = messages.get(e.code, e.description or e.name) + return render_template( + "error.html", code=e.code, name=e.name, text=text + ), e.code + + app.register_blueprint(uploads.bp) + app.register_blueprint(review.bp) + return app + + +def _slug(s: str) -> bool: + try: + Ref(s, "x") + except ValueError: + return False + return True + + +def main(argv: list[str] | None = None) -> int: + import argparse + + from egomimic.utils.env import load_env + + ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0]) + accounts.add_store_args(ap) + ap.add_argument("--host", default="127.0.0.1") + ap.add_argument("--port", type=int, default=8000) + args = ap.parse_args(argv) + if not args.local_store: + load_env() + secret = os.environ.get("EGOVERSE_DASHBOARD_SECRET", "") + if not secret: + raise SystemExit("set EGOVERSE_DASHBOARD_SECRET (e.g. python -c " + "'import secrets; print(secrets.token_urlsafe(48))')") # fmt: skip + insecure = os.environ.get("EGOVERSE_DASHBOARD_INSECURE") == "1" + proxies = int(os.environ.get("EGOVERSE_DASHBOARD_PROXIES", "0")) + app = create_app( + accounts.store_from_args(args), + secret, + secure_cookies=not insecure, + proxies=proxies, + ) + app.run(host=args.host, port=args.port) + return 0 diff --git a/egomimic/ingest/dashboard/review.py b/egomimic/ingest/dashboard/review.py new file mode 100644 index 000000000..cbf226698 --- /dev/null +++ b/egomimic/ingest/dashboard/review.py @@ -0,0 +1,137 @@ +"""Staff review: approve or decline converted deliveries, one at a time or in bulk. + +A decision names the stage (the exact converted result) the reviewer saw, so it only +ever applies to that result: the watcher commits an approved stage on its next pass, and +ignores decisions for a stage that has since been replaced. A decision is refused here +when the delivery changed after the page was loaded (re-uploaded, a correction under +way, re-converted), so staff never approve something they did not see. + + POST /d///decide decision, stage, comment + POST /decide decision, comment, item = vendor/id/stage ... + +Each bulk item is checked on its own; one that changed is skipped and reported. The +newest decision for a stage wins, so a reviewer can change their mind until the watcher +acts. A decline needs a comment: the vendor sees it (not who wrote it). +""" + +from __future__ import annotations + +import re + +from flask import ( + Blueprint, + abort, + current_app, + flash, + g, + redirect, + request, + url_for, +) + +from egomimic.ingest.store import Ref, Store + +MAX_COMMENT = 1000 +MAX_BULK = 500 +_ITEM = re.compile(r"([^/]+)/([^/]+)/([0-9a-f]{64})") + +bp = Blueprint("review", __name__) + + +def _store() -> Store: + return current_app.extensions["ingest_store"] + + +def _staff() -> None: + if g.user is None: + abort(401) + if g.account["role"] != "staff": + abort(404) # vendors do not review; the page does not exist for them + + +def decide(ref: Ref, decision: str, stage: str, comment: str) -> str | None: + """Record a decision on the stage the reviewer saw; the reason it was refused, or + None. Checked against the bucket as it is now.""" + store = _store() + status = store.get_status(ref) or {} + current = store.get_delivery(ref) + if status.get("state") != "ready" or status.get("stage") != stage: + return "it is no longer waiting for review with the result you saw" + if current is None or not current.complete: + return "an upload to it is under way" + if current.fingerprint != status.get("fingerprint"): + return "it was re-uploaded since it was converted" + store.add_decision(ref, decision, g.user, comment, stage) + return None + + +def _form() -> tuple[str, str]: + decision = request.form.get("decision", "") + comment = request.form.get("comment", "").strip() + if decision not in ("approve", "decline"): + abort(400, "decision must be approve or decline") + if len(comment) > MAX_COMMENT: + abort(400, f"comments are at most {MAX_COMMENT} characters") + return decision, comment + + +@bp.post("/d///decide") +def decide_one(vendor, delivery_id): + _staff() + try: + ref = Ref(vendor, delivery_id) + except ValueError: + abort(404) + decision, comment = _form() + page = url_for("delivery", vendor=vendor, delivery_id=delivery_id) + if decision == "decline" and not comment: + flash("A decline needs a comment: the vendor sees it.", "error") + return redirect(page) + stage = request.form.get("stage", "") + if not re.fullmatch(r"[0-9a-f]{64}", stage): + abort(400, "stage missing") + refused = decide(ref, decision, stage, comment) + if refused: + flash(f"Not recorded: {refused}. Reload and review it again.", "error") + else: + done = "Approved" if decision == "approve" else "Declined" + flash(f"{done}; the next pipeline pass acts on it.", "ok") + return redirect(page) + + +@bp.post("/decide") +def decide_many(): + _staff() + decision, comment = _form() + back = request.form.get("back", "") + if not back.startswith("/") or back.startswith("//"): + back = url_for("index") + items = request.form.getlist("item") + if not items: + flash("Tick the deliveries to decide on first.", "error") + return redirect(back) + if len(items) > MAX_BULK: + abort(400, f"at most {MAX_BULK} at once") + if decision == "decline" and not comment: + flash("A decline needs a comment: the vendors see it.", "error") + return redirect(back) + done, skipped = 0, [] + for item in items: + m = _ITEM.fullmatch(item) + try: + ref = Ref(m.group(1), m.group(2)) if m else None + except ValueError: + ref = None + if ref is None: + abort(400, "malformed item") + refused = decide(ref, decision, m.group(3), comment) + if refused: + skipped.append(f"{ref.vendor}/{ref.delivery_id} ({refused})") + else: + done += 1 + verb = "Approved" if decision == "approve" else "Declined" + flash(f"{verb} {done}; the next pipeline pass acts on them.", "ok") + if skipped: + flash(f"Skipped {len(skipped)}, changed since the page loaded: " + + "; ".join(skipped), "error") # fmt: skip + return redirect(back) diff --git a/egomimic/ingest/dashboard/templates.py b/egomimic/ingest/dashboard/templates.py new file mode 100644 index 000000000..3f92428b9 --- /dev/null +++ b/egomimic/ingest/dashboard/templates.py @@ -0,0 +1,524 @@ +"""The dashboard's pages, kept in code rather than separate HTML files. Names end +in .html, which turns on Jinja autoescaping: every vendor-supplied string is escaped.""" + +CSS = """ +:root { --fg: #1d1f23; --muted: #626873; --line: #dde0e5; --bg: #ffffff; + --panel: #f6f7f9; --accent: #2457c5; --ok: #1b7a3d; --warn: #9a5b00; --bad: #b42318; } +@media (prefers-color-scheme: dark) { + :root { --fg: #e6e8eb; --muted: #9aa1ab; --line: #343a42; --bg: #15181c; + --panel: #1d2126; --accent: #7aa2ff; --ok: #5cc98a; --warn: #e3a54a; --bad: #ff7b70; } +} +* { box-sizing: border-box; } +body { margin: 0; font: 15px/1.5 system-ui, sans-serif; color: var(--fg); + background: var(--bg); } +header { display: flex; align-items: center; gap: 16px; padding: 12px 16px; + border-bottom: 1px solid var(--line); } +header .who { margin-left: auto; color: var(--muted); } +header form { display: inline; } +main { max-width: 1100px; margin: 0 auto; padding: 16px; } +a { color: var(--accent); } +h1 { font-size: 20px; margin: 8px 0 16px; } +h2 { font-size: 16px; margin: 24px 0 8px; } +table { width: 100%; border-collapse: collapse; } +th, td { text-align: left; padding: 6px 8px; border-bottom: 1px solid var(--line); + vertical-align: top; } +th { color: var(--muted); font-weight: 500; } +.scroll { overflow-x: auto; } +.num { text-align: right; font-variant-numeric: tabular-nums; } +.muted { color: var(--muted); } +.state { white-space: nowrap; font-weight: 500; } +.state.committed, .state.ready, .state.approved { color: var(--ok); } +.state.rejected, .state.failed, .state.unsupported, .state.declined { color: var(--bad); } +.state.uploading, .state.queued, .state.converting, +.state.commit_rejected { color: var(--warn); } +.panel { background: var(--panel); border: 1px solid var(--line); border-radius: 6px; + padding: 12px 16px; } +ul.problems { margin: 0; padding-left: 20px; } +pre { white-space: pre-wrap; word-break: break-word; margin: 0; } +.filters { display: flex; flex-wrap: wrap; gap: 8px; margin-bottom: 12px; } +input, select, button { font: inherit; padding: 6px 8px; border: 1px solid var(--line); + border-radius: 4px; background: var(--bg); color: var(--fg); } +button { cursor: pointer; } +.login { max-width: 360px; margin: 64px auto; display: grid; gap: 12px; } +.error { color: var(--bad); } +.flash { padding: 8px 12px; border-radius: 4px; border: 1px solid var(--line); } +.flash.ok { color: var(--ok); } +.review { display: flex; flex-wrap: wrap; gap: 8px; align-items: center; margin-top: 12px; } +.review input[name=comment] { flex: 1 1 280px; } +.upload { display: grid; gap: 12px; max-width: 720px; } +.button { display: inline-block; padding: 6px 12px; border: 1px solid var(--accent); + border-radius: 4px; text-decoration: none; } +progress { width: 100%; } +/* Shared visual language: quiet surfaces, clear hierarchy, visible focus. */ +:root { --bg: #f5f7fa; --panel: #fff; --fg: #182b36; --muted: #586b78; + --line: #dce4e9; --accent: #176c62; --ok: #176c62; --warn: #8b5c16; + --bad: #ad3542; --soft: #edf3f5; } +@media (prefers-color-scheme: dark) { + :root { --bg: #111b22; --panel: #19262f; --fg: #e4edf2; --muted: #a4b6c2; + --line: #34454f; --accent: #81d5be; --ok: #81d5be; --warn: #edc078; + --bad: #ffa3ad; --soft: #22333e; } +} +body { line-height: 1.6; } +header { background: var(--panel); padding: 16px max(24px, calc((100vw - 1200px) / 2)); + flex-wrap: wrap; gap: 20px; } +.brand { display: flex; align-items: center; gap: 10px; font-size: 18px; + letter-spacing: -.5px; text-decoration: none; color: var(--fg); font-weight: 750; } +.brand-mark { width: 30px; height: 30px; display: grid; place-items: center; + border-radius: 9px; background: var(--accent); color: var(--panel); font-size: 16px; } +header nav { display: flex; gap: 20px; } +header nav a { color: var(--muted); text-decoration: none; font-weight: 550; } +header nav a[aria-current=page] { color: var(--accent); } +header .who { font-size: 13px; } +main { max-width: 1248px; padding: 40px 24px 72px; } +h1 { font-size: clamp(25px, 4vw, 34px); line-height: 1.2; letter-spacing: -1px; + margin: 6px 0 12px; overflow-wrap: anywhere; } +h2 { font-size: 19px; letter-spacing: -.3px; margin-top: 32px; } +.eyebrow { color: var(--accent); text-transform: uppercase; font-size: 11px; + font-weight: 750; letter-spacing: 1.6px; margin: 0 0 8px; } +.page-head { display: flex; align-items: center; justify-content: space-between; + gap: 20px; margin-bottom: 28px; } +.page-head p { margin: 0; } +.summary { display: grid; grid-template-columns: repeat(4, minmax(0, 1fr)); + gap: 14px; margin-bottom: 24px; } +.metric { background: var(--panel); border: 1px solid var(--line); border-radius: 12px; + padding: 18px 22px; } +.metric strong { display: block; font-size: 28px; line-height: 1.3; font-weight: 650; } +.metric span { color: var(--muted); font-size: 13px; } +.panel, .scroll, .login, .upload { border: 1px solid var(--line); border-radius: 12px; + background: var(--panel); } +.panel { padding: 20px 24px; overflow-wrap: anywhere; } +.scroll { overflow-x: auto; } +th, td { padding: 14px 16px; } +th { font-size: 12px; background: var(--soft); } +td { font-size: 14px; overflow-wrap: anywhere; } +tr:last-child td { border-bottom: 0; } +tr:hover td { background: var(--soft); } +td a { font-weight: 600; text-decoration: none; } +.state { display: inline-block; border: 1px solid currentColor; border-radius: 6px; + padding: 2px 8px; font-size: 12px; font-weight: 650; } +.filters { gap: 10px; margin-bottom: 18px; align-items: center; } +input, select, button, .button { border-radius: 7px; padding: 9px 12px; } +button, .button { font-weight: 600; } +button:hover, .button:hover { border-color: var(--accent); } +button:disabled { cursor: wait; opacity: .6; } +button[value=approve], .primary, .login button, #go { background: var(--accent); + border-color: var(--accent); color: var(--panel); } +button[value=decline] { color: var(--bad); } +input[type=checkbox] { width: 17px; height: 17px; accent-color: var(--accent); } +:focus-visible { outline: 3px solid var(--accent); outline-offset: 3px; } +.review { padding: 16px; background: var(--panel); border: 1px solid var(--line); + border-radius: 10px; } +.login { padding: 32px; max-width: 420px; margin: 48px auto; gap: 18px; } +.login input { width: 100%; margin-top: 5px; } +.upload { max-width: 850px; padding: 24px; gap: 20px; } +.upload input[type=file] { display: block; width: 100%; margin-top: 8px; + padding: 18px; border-style: dashed; } +.upload label { font-weight: 600; } +.upload .panel { background: var(--soft); } +.empty { text-align: center; padding: 48px 24px; } +.empty h2 { margin: 0 0 8px; } +.flash { background: var(--panel); padding: 14px 18px; } +code { font-size: .88em; background: var(--soft); padding: 2px 5px; border-radius: 4px; + overflow-wrap: anywhere; } +.reference { display: grid; grid-template-columns: 220px minmax(0, 1fr); gap: 36px; } +.reference nav { position: sticky; top: 24px; align-self: start; display: grid; gap: 8px; } +.reference nav a { text-decoration: none; color: var(--muted); font-size: 14px; } +.reference section { scroll-margin-top: 24px; margin-bottom: 28px; } +.reference section > h2 { margin-top: 0; } +.flow { display: grid; grid-template-columns: repeat(5, minmax(0, 1fr)); + gap: 10px; padding: 0; list-style: none; margin: 0 0 28px; counter-reset: step; } +.flow li { padding: 16px; background: var(--panel); border: 1px solid var(--line); + border-radius: 10px; counter-increment: step; } +.flow li::before { content: '0' counter(step); display: block; color: var(--accent); + font-size: 12px; font-weight: 700; margin-bottom: 8px; } +.flow small { display: block; color: var(--muted); } +@media (max-width: 700px) { + header { padding: 14px 16px; gap: 12px; } + header .who { margin-left: 0; } + header nav { gap: 12px; } + main { padding: 24px 16px 48px; } + .page-head { align-items: flex-start; flex-direction: column; } + .summary { grid-template-columns: repeat(2, minmax(0, 1fr)); } + .metric { padding: 14px; } + .reference { grid-template-columns: 1fr; gap: 24px; } + .reference nav { position: static; grid-template-columns: repeat(2, minmax(0, 1fr)); } + .flow { grid-template-columns: 1fr; } + .login { margin: 20px auto; padding: 24px; } + .filters select { max-width: 100%; } + .upload { padding: 18px; } +} +""" + +BASE = """ + + +{% block title %}EgoVerse deliveries{% endblock %} + + +
EgoVerse / deliveries +{% if user %} +{{ user }} · {{ "staff" if staff else account.vendor }} +
+
+{% endif %}
+
{% for kind, text in get_flashed_messages(with_categories=true) %} +

{{ text }}

{% endfor %} +{% block body %}{% endblock %}
+""" + +LOGIN = """{% extends "base.html" %} +{% block title %}Log in · EgoVerse deliveries{% endblock %} +{% block body %} + +{% endblock %}""" + +INDEX = """{% extends "base.html" %} +{% block body %} +

{{ 'Review workspace' if staff else 'Vendor workspace' }}

+

{{ vendor or "All vendors" }}: deliveries

+

{{ 'Follow incoming recordings from upload to reviewed dataset.' if staff else 'Send recordings and track their progress into the dataset.' }}

+{% if not staff %}New delivery{% endif %}
+
+{% for label, value in summary %}
{{ value }}{{ label }}
{% endfor %} +
+

Summary covers all states{{ ' for ' ~ vendor if vendor else ' across all vendors' }}; filters below narrow the table.

+
+{% if staff %}{% endif %} +
+{% if not rows %}

{{ 'No matching deliveries' if state else 'No deliveries yet.' }}

+

{{ 'Choose another state to see more recordings.' if state else 'Deliveries will appear here once files arrive.' }}

{% else %} +{% if staff %}
+ +{% endif %} +
+{% if staff %}{% endif %} + + +{% for r in rows %} +{% if staff %} +{% endif %} + + + + + + +{% endfor %} +
VendorDeliveryStateSizeFramesProblemsUpdated (UTC)
{% if r.decidable %}{% endif %}{{ r.ref.vendor }}{{ r.ref.delivery_id }}{{ r.state }}{{ r.size | size }}{{ r.frames if r.frames is not none else "" }}{{ r.problems | length or "" }}{{ r.updated | when }}
+{% if staff %}
+ + + +
{% endif %} +{% endif %} +{% endblock %}""" + +DELIVERY = """{% extends "base.html" %} +{% block title %}{{ r.ref.delivery_id }} · EgoVerse deliveries{% endblock %} +{% block body %} +

Delivery details

+

{{ r.ref.vendor }} / {{ r.ref.delivery_id }}

+{% if not staff %}

+Upload a correction

{% endif %} +
{{ r.state }}: +{{ STATES.get(r.state, "") }} +{% if r.status.recording_id %}
Recording +{{ r.status.recording_id }}{% endif %} +{% if r.status.format %}
Format +{{ r.status.format }}{% endif %} +{% if decision %}
Decision {{ decision.decision }} +{% if decision.comment %}: {{ decision.comment }}{% endif %} +{% if decision.who %}by {{ decision.who }}{% endif %} +{% endif %}
+ +{% if staff and r.decidable %}

Review

+
+ + +Result {{ r.status.stage[:12] }}{% if r.decision %}: currently +{{ r.decision.decision }}d by {{ r.decision.who }}{% endif %} + + + +
{% endif %} + +{% if r.problems %}

Problems ({{ r.problems | length }})

+
    +{% for p in r.problems %}
  • {{ p }}
    • {% endfor %}
{% endif %} + +{% if error %}

Error

{{ error }}
{% endif %} + +{% if r.status.report %}

Conversion

+{% for k, v in r.status.report.items() %}{% endfor %} +
{{ k }} +{% if v is mapping %}{% for kk, vv in v.items() %}{{ kk }}: {{ vv | fmt }}{{ ", " if not loop.last }} +{% endfor %}{% else %}{{ v | fmt }}{% endif %}
{% endif %} + +{% if r.status.diff %}

What changes in the dataset

+{% for k, v in r.status.diff.items() %}{% endfor %} +
{{ k }} +{% if v is mapping %}{% for kk, vv in v.items() %}{{ kk }}: {{ vv | fmt }}
+{% endfor %}{% else %}{{ v | fmt }}{% endif %}
{% endif %} + +

Files

{% if r.files %}
+ +{% for name, (size, etag) in r.files | dictsort %} +{% endfor %} +
NameSize
{{ name }}{{ size | size }}
{% else %}

No files.

{% endif %} + +{% if staff and r.status.provenance %}

Made by

+{% for k, v in r.status.provenance.items() %} +{% endfor %}
{{ k }}{{ v | fmt }}
{% endif %} + +{% if r.status.history %}

History

+{% for h in r.status.history | reverse %} + +{% endfor %} +
{{ h.state }}{{ h.at | when }}
{% endif %} + +{% if staff and decisions %}

Decisions

+ +{% for d in decisions | reverse %} + +{% endfor %} +
DecisionByCommentStageAt (UTC)
{{ d.decision }}{{ d.who }}{{ d.comment }}{{ (d.stage or "")[:12] }}{{ d.at | when }}
{% endif %} +{% endblock %}""" + +NEW = """{% extends "base.html" %} +{% block title %}New delivery · EgoVerse deliveries{% endblock %} +{% block body %} +

{{ "Correct " + delivery_id if delivery_id else "New delivery" }}

+
+ +

Letters, digits, . _ -; one +recording per delivery. Uploading to an existing name replaces its files: send the whole +delivery again, not only what changed.

+ + +
Accepted +
    {% for f in formats %}
  • {{ f.name }}: +{{ f.required | sort | join(", ") }}{% if f.optional %} (optional: +{{ f.optional | sort | join(", ") }}){% endif %}. {{ f.description }}
    • {% endfor %} +
At most {{ limit_gib }} GiB per file.
+
+ +

Large deliveries or scripts: the +command-line uploader (Python 3.10+, nothing +to install): python egoverse_upload.py --url {{ request.host_url.rstrip("/") }} +--user {{ user }} <folder>

+ +

+
+ +{% endblock %}""" + +UPLOAD_JS = """'use strict'; +// Upload a delivery: each file goes straight to the bucket in equal parts through +// presigned links (the dashboard never sees the bytes), then the delivery is marked +// complete as exactly these files. See egomimic/ingest/dashboard/uploads.py. +(() => { + const form = document.getElementById('upload'); + if (!form) return; + const csrf = form.dataset.csrf; + const $ = (id) => document.getElementById(id); + const PARALLEL = 4, RETRIES = 4, BATCH = 100; + let busy = false; + + const say = (text, bad) => { $('msg').textContent = text; $('msg').className = bad ? 'error' : ''; }; + const human = (n) => { const u = ['B', 'KB', 'MB', 'GB', 'TB']; let i = 0; + while (n >= 1000 && i < u.length - 1) { n /= 1000; i++; } return (i ? n.toFixed(1) : n) + ' ' + u[i]; }; + + async function api(method, path, body) { + const r = await fetch(path, { method, credentials: 'same-origin', + headers: { 'Content-Type': 'application/json', 'X-CSRF-Token': csrf }, + body: body === undefined ? undefined : JSON.stringify(body) }); + const data = await r.json().catch(() => ({})); + if (!r.ok) { const e = new Error(data.error || r.status + ' ' + r.statusText); e.data = data; throw e; } + return data; + } + + // Relative names: drop the picked folder's own name, keep subfolders. + function picked() { + const folder = [...form.folder.files], files = [...form.files.files]; + const list = folder.length ? folder : files; + return list.map((f) => ({ file: f, + name: f.webkitRelativePath ? f.webkitRelativePath.split('/').slice(1).join('/') : f.name })); + } + + function show() { + const t = $('picked'); t.replaceChildren(); + for (const { file, name } of picked()) { + const tr = t.insertRow(); tr.insertCell().textContent = name; + const td = tr.insertCell(); td.className = 'num'; td.textContent = human(file.size); + } + } + form.folder.addEventListener('change', show); + form.files.addEventListener('change', show); + + async function retry(fn) { + for (let i = 0; ; i++) { + try { return await fn(); } catch (e) { + if (i + 1 >= RETRIES) throw e; + await new Promise((ok) => setTimeout(ok, 1000 * 2 ** i)); + } + } + } + + async function putPart(base, name, start, n, url, blob) { + if (url) { // presigned: straight to the bucket; its CORS must expose ETag + const r = await fetch(url, { method: 'PUT', body: blob }); + if (!r.ok) throw new Error('part ' + n + ': ' + r.status); + const etag = r.headers.get('ETag'); + if (!etag) throw new Error('the bucket does not expose ETag (CORS)'); + return etag; + } + const q = new URLSearchParams({ name, upload_id: start.upload_id, part: n }); + const r = await fetch(base + '/part?' + q, { method: 'PUT', body: blob, + credentials: 'same-origin', headers: { 'X-CSRF-Token': csrf } }); + const data = await r.json().catch(() => ({})); + if (!r.ok) throw new Error(data.error || 'part ' + n + ': ' + r.status); + return data.etag; + } + + async function uploadFile(base, { file, name }, progress) { + const start = await api('POST', base + '/uploads', { name, size: file.size }); + const ref = { name, upload_id: start.upload_id }; + try { + const etags = []; + for (let first = 1; first <= start.parts; first += BATCH) { + const nums = []; + for (let n = first; n < Math.min(first + BATCH, start.parts + 1); n++) nums.push(n); + const { urls } = await api('POST', base + '/urls', { ...ref, parts: nums }); + const queue = [...nums]; + const worker = async () => { + for (let n; (n = queue.shift()) !== undefined;) { + const blob = file.slice((n - 1) * start.part_size, n * start.part_size); + etags.push([n, await retry(() => putPart(base, name, start, n, urls[n], blob))]); + progress(blob.size); + } + }; + await Promise.all(Array.from({ length: PARALLEL }, worker)); + } + await api('POST', base + '/finish', { ...ref, parts: etags.sort((a, b) => a[0] - b[0]) }); + } catch (e) { + await api('POST', base + '/abort', ref).catch(() => {}); + throw e; + } + } + + form.addEventListener('submit', async (ev) => { + ev.preventDefault(); + if (busy) return; + const id = form.delivery.value.trim(), files = picked(); + if (!files.length) return say('Pick a folder or files first.', true); + const base = '/api/deliveries/' + encodeURIComponent(id); + busy = true; $('go').disabled = true; + const total = files.reduce((n, f) => n + f.file.size, 0) || 1; + let done = 0; + const bar = $('bar'); bar.hidden = false; bar.value = 0; + window.onbeforeunload = () => 'The upload is still running.'; + try { + const now = await api('GET', base); + if (now.uploads.length) { + if (!confirm(now.uploads.length + ' unfinished upload(s) to ' + id + + ' (e.g. a closed tab). Drop them and continue?')) throw new Error('Stopped.'); + for (const u of now.uploads) await api('POST', base + '/clear', { record: u.record }); + } + for (const [i, f] of files.entries()) { + say('Uploading ' + f.name + ' (' + (i + 1) + ' of ' + files.length + ')'); + await uploadFile(base, f, (n) => { done += n; bar.value = done / total; }); + } + say('Finishing...'); + await api('POST', base + '/complete', { files: files.map((f) => f.name) }); + window.onbeforeunload = null; + location.href = '/d/' + now.vendor + '/' + encodeURIComponent(id); + } catch (e) { + say('Upload failed: ' + e.message, true); + } finally { + busy = false; $('go').disabled = false; window.onbeforeunload = null; + } + }); +})(); +""" + +REVIEW_JS = """'use strict'; +// Staff review: the header checkbox ticks every delivery waiting for review. +(() => { + const all = document.getElementById('all'); + if (!all) return; + const picks = () => [...document.querySelectorAll('input.pick')]; + all.addEventListener('change', () => picks().forEach((c) => { c.checked = all.checked; })); +})(); +""" + +ERROR = """{% extends "base.html" %} +{% block title %}{{ name }} · EgoVerse deliveries{% endblock %} +{% block body %} +

{{ name }}

+

{{ text }}

+

Back to +{{ "the deliveries" if user else "log in" }}

+{% endblock %}""" + + +TEMPLATES = { + "base.html": BASE, + "login.html": LOGIN, + "index.html": INDEX, + "delivery.html": DELIVERY, + "new.html": NEW, + "error.html": ERROR, +} + + +def size(n: int) -> str: + for unit in ("B", "KB", "MB", "GB", "TB"): + if n < 1000 or unit == "TB": + return f"{n:.0f} {unit}" if unit == "B" else f"{n:.1f} {unit}" + n /= 1000 + return "" + + +def when(at: str) -> str: + """store.now() stamps, 20261021T140312.123456Z -> 2026-10-21 14:03""" + if len(at) < 13: + return at or "" + return f"{at[:4]}-{at[4:6]}-{at[6:8]} {at[9:11]}:{at[11:13]}" + + +def fmt(v) -> str: + if isinstance(v, bool) or v is None: + return str(v).lower() + if isinstance(v, float): + return f"{v:.3g}" + if isinstance(v, dict): + return ", ".join(f"{k}: {fmt(x)}" for k, x in v.items()) if v else "none" + if isinstance(v, (list, tuple)): + return ", ".join(fmt(x) for x in v) if v else "none" + return str(v) + + +FILTERS = {"size": size, "when": when, "fmt": fmt} diff --git a/egomimic/ingest/dashboard/throttle.py b/egomimic/ingest/dashboard/throttle.py new file mode 100644 index 000000000..5cff2aa78 --- /dev/null +++ b/egomimic/ingest/dashboard/throttle.py @@ -0,0 +1,79 @@ +"""Login throttling: sliding-window counts kept in memory, checked before any password +hashing (scrypt is deliberately expensive, so unthrottled it is a DoS lever). + +Limits are temporary (a window, not a lockout), so an attacker cannot lock a vendor out +for good. Three keys: +- (ip, user) failures: guessing one account from one place; +- ip attempts: one source trying many accounts, or flooding scrypt; +- user failures: a looser cap on one account guessed from many places. + +In-memory, so per process: the dashboard runs as one process (DEPLOY.md), and the AWS +edge limits requests per IP in front of it as well. +""" + +from __future__ import annotations + +import threading +import time +from collections import deque + +WINDOW = 15 * 60 # seconds +PAIR_FAILURES = 5 +IP_ATTEMPTS = 30 +USER_FAILURES = 50 +MAX_KEYS = 100_000 # memory bound; the oldest keys go first + + +class Throttle: + def __init__(self, clock=time.monotonic): + self._clock = clock + self._hits: dict[tuple, deque] = {} + self._lock = threading.Lock() + + def _count(self, key: tuple, now: float) -> deque: + hits = self._hits.get(key) + if hits is None: + hits = self._hits[key] = deque() + while hits and hits[0] <= now - WINDOW: + hits.popleft() + return hits + + def retry_after(self, ip: str, user: str) -> int: + """Seconds until this login may be tried, 0 = go ahead (and count it).""" + with self._lock: + now = self._clock() + waits = [] + for key, limit in ( + (("pair", ip, user), PAIR_FAILURES), + (("ip", ip), IP_ATTEMPTS), + (("user", user), USER_FAILURES), + ): + hits = self._count(key, now) + if len(hits) >= limit: + waits.append(hits[0] + WINDOW - now) + if waits: + return max(1, int(max(waits) + 0.999)) + self._count(("ip", ip), now).append(now) # every attempt counts per ip + self._prune() + return 0 + + def failed(self, ip: str, user: str) -> None: + with self._lock: + now = self._clock() + self._count(("pair", ip, user), now).append(now) + self._count(("user", user), now).append(now) + self._prune() + + def succeeded(self, ip: str, user: str) -> None: + with self._lock: + self._hits.pop(("pair", ip, user), None) + + def _prune(self) -> None: + if len(self._hits) > MAX_KEYS: + now = self._clock() + for key in [ + k for k, v in self._hits.items() if not v or v[-1] <= now - WINDOW + ]: + del self._hits[key] + while len(self._hits) > MAX_KEYS: + del self._hits[next(iter(self._hits))] diff --git a/egomimic/ingest/dashboard/uploads.py b/egomimic/ingest/dashboard/uploads.py new file mode 100644 index 000000000..c667f7caa --- /dev/null +++ b/egomimic/ingest/dashboard/uploads.py @@ -0,0 +1,301 @@ +"""Uploads: a vendor sends a delivery's files straight to the bucket, in parts, through +presigned links the dashboard hands out for that vendor's own folder only. No bucket +credentials leave the lab, and the file bytes never pass through the dashboard (except +on the local-directory store, which has no presigning: there parts come through here). + +The JSON API, shared by the upload page (upload.js) and the CLI (egomimic.ingest.upload); +every POST carries the session's CSRF token in X-CSRF-Token: + + POST /api/login {user, password} -> {csrf} + GET /api/session -> {user, role, vendor, csrf} + GET /api/deliveries/ -> {state, files, uploads} + POST /api/deliveries//uploads {name, size} -> {upload_id, part_size, parts} + POST /api/deliveries//urls {name, upload_id, parts: [n]} -> {urls: {n: url}} + PUT /api/deliveries//part ?name&upload_id&part (local store only) -> {etag} + POST /api/deliveries//finish {name, upload_id, parts: [[n, etag]]} + POST /api/deliveries//abort {name, upload_id} + POST /api/deliveries//clear {record}: an abandoned upload + POST /api/deliveries//complete {files: [names]}: the delivery is exactly these + +A delivery is processed only once marked complete; marking it complete with a file list +deletes any other file in it, so a correction is a full re-upload of the folder. +""" + +from __future__ import annotations + +import math + +from flask import Blueprint, abort, current_app, g, jsonify, request, session + +from egomimic.ingest import formats +from egomimic.ingest.dashboard import views +from egomimic.ingest.store import ( + Ref, + Store, + UploadRejected, + UploadsInProgress, + _file_name, + part_length, +) + +MIN_PART = ( + 16 * 1024**2 +) # R2: parts of one upload are equal-sized, >= 5 MiB but the last +MAX_PARTS = 10_000 +URLS_PER_CALL = 100 +URL_EXPIRES_S = 3600 +MAX_FILES = 1000 +JSON_LIMIT = 2 * 1024**2 # a finish call lists up to 10k (part, etag) pairs + +bp = Blueprint("uploads", __name__) + + +def _store() -> Store: + return current_app.extensions["ingest_store"] + + +def _fail(status: int, message: str, **extra): + resp = jsonify({"error": message, **extra}) + resp.status_code = status + abort(resp) + + +def _body() -> dict: + request.max_content_length = JSON_LIMIT + body = request.get_json(silent=True) + if not isinstance(body, dict): + _fail(400, "expected a JSON object") + return body + + +def _ref(delivery_id: str) -> Ref: + """The signed-in vendor's own delivery; staff do not upload.""" + if g.user is None: + _fail(401, "log in first") + if g.account["role"] != "vendor": + _fail(403, "only vendor accounts upload") + try: + return Ref(g.account["vendor"], delivery_id) + except ValueError as e: + _fail(400, str(e)) + + +def _name(body: dict) -> str: + try: + return _file_name(str(body.get("name", ""))) + except ValueError as e: + _fail(400, str(e)) + + +def _upload(ref: Ref, body: dict) -> tuple[str, str, dict]: + """(name, upload id, record) of an unfinished upload to this delivery, as recorded + when it started: an id from another delivery or file is refused. Every upload + started here is size-bound (size, part size, part count in the record).""" + name, upload_id = _name(body), str(body.get("upload_id", "")) + rec = _store().upload_record(ref, upload_id) if upload_id else None + if not rec or rec.get("name") != name or rec.get("size") is None: + _fail(404, "no such upload in progress") + return name, upload_id, rec + + +def _length(rec: dict, n) -> int: + if not isinstance(n, int): + _fail(400, "part numbers are integers") + try: + return part_length(rec, n) + except UploadRejected as e: + _fail(400, str(e)) + + +def accepted() -> tuple[frozenset[str], int]: + names = frozenset().union(*(f.allowed for f in formats.formats())) + return names, max(f.max_file_bytes for f in formats.formats()) + + +def part_size(size: int) -> int: + """Equal parts (R2 requires it), at most MAX_PARTS of them, in whole MiB.""" + need = math.ceil(size / MAX_PARTS) + if need <= MIN_PART: + return MIN_PART + return math.ceil(need / 1024**2) * 1024**2 + + +@bp.post("/api/login") +def api_login(): + """Log in for the CLI: the same checks and throttling as the login form.""" + if not request.is_json: # a cross-site form cannot send JSON without a preflight + _fail(415, "send JSON") + body = _body() + ok, error, status = current_app.extensions["ingest_login"]( + str(body.get("user", "")), str(body.get("password", "")) + ) + if not ok: + _fail(status, error) + return jsonify({"csrf": session["csrf"]}) + + +@bp.get("/api/session") +def api_session(): + if g.user is None: + _fail(401, "log in first") + a = g.account + return jsonify( + { + "user": g.user, + "role": a["role"], + "vendor": a["vendor"], + "csrf": session["csrf"], + } + ) + + +@bp.get("/api/deliveries/") +def api_delivery(delivery_id): + ref = _ref(delivery_id) + store = _store() + r = views.row(store, ref) + return jsonify( + { + "vendor": ref.vendor, + "state": r.state if r else "new", + "files": {n: size for n, (size, _) in (r.files if r else {}).items()}, + "uploads": [ + {k: u.get(k) for k in ("record", "name", "state", "started_at")} + for u in store.uploads_in_progress(ref) + ], + } + ) + + +@bp.post("/api/deliveries//uploads") +def api_start(delivery_id): + ref = _ref(delivery_id) + body = _body() + name = _name(body) + size = body.get("size") + if not isinstance(size, int) or size < 0: + _fail(400, "size must be the file's size in bytes") + allowed, limit = accepted() + if name not in allowed: + _fail(422, f"{name}: no accepted format has this file; expected one of " + f"{', '.join(sorted(allowed))}") # fmt: skip + if size > limit: + _fail(422, f"{name}: {size} bytes is over the {limit / 1024**3:.0f} GiB limit") + ps = part_size(size) + store = _store() + upload_id = store.start_upload(ref, name, size=size, part_size=ps) + rec = store.upload_record(ref, upload_id) + return jsonify( + {"upload_id": upload_id, "part_size": rec["part_size"], "parts": rec["parts"]} + ) + + +@bp.post("/api/deliveries//urls") +def api_urls(delivery_id): + ref = _ref(delivery_id) + body = _body() + name, upload_id, rec = _upload(ref, body) + parts = body.get("parts") + if not isinstance(parts, list) or not 0 < len(parts) <= URLS_PER_CALL: + _fail(400, f"parts: 1-{URLS_PER_CALL} part numbers") + lengths = {n: _length(rec, n) for n in parts} # only 1..the declared part count + store = _store() + urls = { + str(n): store.part_url(ref, name, upload_id, n, URL_EXPIRES_S, length) + for n, length in lengths.items() + } # each signed for its exact length + return jsonify({"urls": urls}) # null: PUT the part to /part (local store) + + +@bp.put("/api/deliveries//part") +def api_part(delivery_id): + """Parts through the dashboard: only for the local store, which cannot presign.""" + ref = _ref(delivery_id) + store = _store() + args = { + "name": request.args.get("name", ""), + "upload_id": request.args.get("upload_id", ""), + } + name, upload_id, rec = _upload(ref, args) + if store.part_url(ref, name, upload_id, 1) is not None: + _fail(404, "parts go to the presigned links") + try: + part = int(request.args.get("part", "")) + except ValueError: + _fail(400, "part must be a number") + length = _length(rec, part) + request.max_content_length = rec["part_size"] + data = request.get_data() + if len(data) != length: + _fail(400, f"part {part} must be {length} bytes, got {len(data)}") + etag = store.upload_part(ref, name, upload_id, part, data) + return jsonify({"etag": etag}) + + +@bp.post("/api/deliveries//finish") +def api_finish(delivery_id): + ref = _ref(delivery_id) + body = _body() + name, upload_id, rec = _upload(ref, body) + parts = body.get("parts") + try: + parts = [(int(n), str(etag)) for n, etag in parts] + except (TypeError, ValueError): + _fail(400, "parts: [[part number, etag], ...]") + try: + _store().finish_upload(ref, name, upload_id, sorted(parts)) + except UploadRejected as e: # wrong parts, or the joined file is not the size + _fail(422, str(e)) + except Exception as e: # a missing or mismatched part: the client retries or aborts + _fail(409, f"could not join the parts of {name}: {type(e).__name__}") + return jsonify({"name": name}) + + +@bp.post("/api/deliveries//abort") +def api_abort(delivery_id): + ref = _ref(delivery_id) + name, upload_id, _ = _upload(ref, _body()) + _store().abort_upload(ref, name, upload_id) + return jsonify({"name": name}) + + +@bp.post("/api/deliveries//clear") +def api_clear(delivery_id): + """Drop an abandoned upload (a closed tab, a crashed CLI) so the delivery can be + completed again.""" + ref = _ref(delivery_id) + record = str(_body().get("record", "")) + store = _store() + if record not in {u["record"] for u in store.uploads_in_progress(ref)}: + _fail(404, "no such upload in progress") + store.clear_upload(ref, record) + return jsonify({"cleared": record}) + + +@bp.post("/api/deliveries//complete") +def api_complete(delivery_id): + ref = _ref(delivery_id) + files = _body().get("files") + if not isinstance(files, list) or not 0 < len(files) <= MAX_FILES: + _fail(400, f"files: the delivery's file names (1-{MAX_FILES})") + try: + names = {_file_name(str(f)) for f in files} + except ValueError as e: + _fail(400, str(e)) + try: + fmt = formats.detect(names) + except formats.AmbiguousFormat as e: + _fail(422, str(e)) + if fmt is None or fmt.unexpected(names): + _fail(422, "these files are not a delivery in any accepted format", + formats=[{"name": f.name, "required": sorted(f.required), + "optional": sorted(f.optional)} for f in formats.formats()]) # fmt: skip + store = _store() + try: + done = store.mark_complete(ref, {"by": g.user}, sorted(names)) + except UploadsInProgress as e: + _fail(409, "uploads to this delivery are still in progress", + uploads=[u["record"] for u in e.records]) # fmt: skip + except ValueError as e: # a listed file was never uploaded + _fail(422, str(e)) + return jsonify({"files": {n: size for n, (size, _) in done.items()}}) diff --git a/egomimic/ingest/dashboard/views.py b/egomimic/ingest/dashboard/views.py new file mode 100644 index 000000000..fb0e1045f --- /dev/null +++ b/egomimic/ingest/dashboard/views.py @@ -0,0 +1,138 @@ +"""What the dashboard shows about a delivery: the bucket's listing and the watcher's +status merged into one row, and what of it a vendor may see.""" + +from __future__ import annotations + +import re +from dataclasses import dataclass, field + +from egomimic.ingest.store import Listed, Ref, Store + +# What each state means, in the words a vendor reads. +STATES = { + "uploading": "Upload in progress", + "queued": "Waiting to be checked", + "converting": "Checking and converting", + "rejected": "Rejected: fix the problems and upload again", + "unsupported": "Not a format we accept", + "failed": "Failed on our side; we are looking into it", + "ready": "Converted; waiting for EgoVerse review", + "approved": "Approved; added to the dataset at the next pass", + "committed": "In the dataset", + "commit_rejected": "Approved, but not added yet; EgoVerse is looking into it", + "declined": "Declined by EgoVerse", +} + + +@dataclass +class Row: + ref: Ref + state: str + files: dict = field(default_factory=dict) # name: (size, etag) + status: dict = field(default_factory=dict) + decision: dict = field(default_factory=dict) # the newest, if for the current stage + + @property + def decidable(self) -> bool: + """Staff can approve or decline it: the watcher has it ready (whatever staff + decided since, so a decline can still become an approval before the watcher + acts) and the bucket has not moved on (re-uploaded, a correction under way).""" + return ( + self.status.get("state") == "ready" + and bool(self.status.get("stage")) + and self.state in ("ready", "approved", "declined") + ) + + @property + def key(self) -> str: + return f"{self.ref.vendor}/{self.ref.delivery_id}" + + @property + def size(self) -> int: + return sum(size for size, _ in self.files.values()) + + @property + def updated(self) -> str: + """The status's last write (a retry or an error leaves the state unchanged).""" + history = self.status.get("history") or [] + return self.status.get("updated_at") or (history[-1]["at"] if history else "") + + @property + def frames(self): + return (self.status.get("report") or {}).get("num_frames") + + @property + def problems(self) -> list: + return self.status.get("problems") or [] + + +def state_of( + listed: Listed | None, status: dict | None, decision: dict | None = None +) -> str: + """The watcher's state, unless the bucket has moved on since it wrote it, or staff + decided on its current stage and the watcher has not acted yet.""" + if listed is not None and not listed.complete: + return "uploading" # a first upload, or a correction under way + if status is None: + return "queued" + if listed is not None and status.get("fingerprint") != listed.fingerprint: + return "queued" # re-uploaded: the next pass checks the new files + state = status.get("state", "queued") + if state == "ready" and decision and decision.get("stage") == status.get("stage"): + return "approved" if decision["decision"] == "approve" else "declined" + return state + + +def rows(store: Store, vendor: str | None = None) -> list[Row]: + """Every delivery of one vendor (None: all), newest activity first.""" + listed = {d.ref: d for d in store.list_deliveries(vendor)} + statuses = store.statuses(vendor) + decided = store.latest_decisions(vendor) + out = [] + for ref in listed.keys() | statuses.keys(): + out.append(_row(ref, listed.get(ref), statuses.get(ref), decided.get(ref))) + return sorted(out, key=lambda r: (r.updated, r.key), reverse=True) + + +def row(store: Store, ref: Ref) -> Row | None: + d = store.get_delivery(ref) + st = store.get_status(ref) + if d is None and st is None: + return None + decisions = store.decisions(ref) + return _row(ref, d, st, decisions[-1] if decisions else None) + + +def _row(ref, listed, status, decision) -> Row: + status = status or {} + current = ( + decision if decision and decision.get("stage") == status.get("stage") else None + ) + return Row( + ref, + state_of(listed, status or None, current), + listed.files if listed else {}, + status, + current or {}, + ) + + +_PATH = re.compile(r"(? str: + """The error as a vendor sees it: the first line of what the converter said about + their files, with our paths blanked; nothing of our own crashes (a traceback) or + of commit errors, which are ours to fix.""" + error = (status.get("error") or "").strip() + if not error or status.get("state") not in ("failed", "rejected", "unsupported"): + return "" + if "Traceback" in error: + return "" + return _PATH.sub("", error.splitlines()[0])[:300] + + +def vendor_decision(decision: dict) -> dict: + """The decision without who made it.""" + d = decision or {} + return {k: d[k] for k in ("decision", "comment", "at") if d.get(k)} diff --git a/egomimic/ingest/upload.py b/egomimic/ingest/upload.py new file mode 100644 index 000000000..1b812da90 --- /dev/null +++ b/egomimic/ingest/upload.py @@ -0,0 +1,221 @@ +"""Upload a delivery to the EgoVerse dashboard from the command line (for big folders, +or scripts). Standard library only, so a vendor needs nothing beyond Python 3.10+. + + python egoverse_upload.py --url https:// --user \\ + [--delivery-id ID] + +(egoverse_upload.py is this file, from the dashboard's /upload.py; in the repo it is +python -m egomimic.ingest.upload.) + +The password is asked for, or read from EGOVERSE_UPLOAD_PASSWORD. Every file under the +folder is uploaded (subfolders kept) straight to the bucket in parts through presigned +links, then the delivery is marked complete as exactly those files: uploading to an +existing delivery replaces it. The delivery id defaults to the folder's name. +""" + +from __future__ import annotations + +import argparse +import getpass +import http.cookiejar +import json +import os +import sys +import time +import urllib.error +import urllib.parse +import urllib.request +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path + +PARALLEL = 4 +RETRIES = 4 +BATCH = 100 + + +class UploadError(Exception): + pass + + +class Client: + def __init__(self, url: str, opener=None): + self.url = url.rstrip("/") + self.opener = opener or urllib.request.build_opener( + urllib.request.HTTPCookieProcessor(http.cookiejar.CookieJar()) + ) + self.csrf = "" + + def call(self, method: str, path: str, body=None, *, data=None, query=None): + url = self.url + path + ("?" + urllib.parse.urlencode(query) if query else "") + headers = {"X-CSRF-Token": self.csrf, "Accept": "application/json"} + if body is not None: + data = json.dumps(body).encode() + headers["Content-Type"] = "application/json" + req = urllib.request.Request(url, data=data, method=method, headers=headers) + try: + with self.opener.open(req, timeout=300) as r: + return json.loads(r.read() or b"{}") + except urllib.error.HTTPError as e: + try: + msg = json.loads(e.read()).get("error", "") + except ValueError: + msg = "" + raise UploadError(f"{method} {path}: {e.code} {msg or e.reason}") from None + + def login(self, user: str, password: str) -> None: + self.csrf = self.call( + "POST", "/api/login", {"user": user, "password": password} + )["csrf"] + + +def _retry(fn): + for i in range(RETRIES): + try: + return fn() + except (UploadError, urllib.error.URLError, OSError): + if i + 1 == RETRIES: + raise + time.sleep(2**i) + + +def _put(url: str, data: bytes) -> str: + req = urllib.request.Request(url, data=data, method="PUT") + with urllib.request.urlopen(req, timeout=600) as r: # presigned: no cookies + etag = r.headers.get("ETag") + if not etag: + raise UploadError("the bucket returned no ETag") + return etag # exactly as given + + +def upload_file(c: Client, base: str, path: Path, name: str, progress=None) -> None: + size = path.stat().st_size + start = c.call("POST", f"{base}/uploads", {"name": name, "size": size}) + ref = {"name": name, "upload_id": start["upload_id"]} + ps, n_parts = start["part_size"], start["parts"] + + def part(n: int, url: str | None) -> list: + with open(path, "rb") as f: + f.seek((n - 1) * ps) + data = f.read(ps) + if url: + etag = _retry(lambda: _put(url, data)) + else: # the local-directory store: parts go through the dashboard + q = {**ref, "part": n} + etag = _retry( + lambda: c.call("PUT", f"{base}/part", data=data, query=q)["etag"] + ) + if progress: + progress(len(data)) + return [n, etag] + + try: + etags = [] + with ThreadPoolExecutor(PARALLEL) as pool: + for first in range(1, n_parts + 1, BATCH): + nums = list(range(first, min(first + BATCH, n_parts + 1))) + urls = c.call("POST", f"{base}/urls", {**ref, "parts": nums})["urls"] + etags += pool.map(lambda n: part(n, urls[str(n)]), nums) + c.call("POST", f"{base}/finish", {**ref, "parts": sorted(etags)}) + except BaseException: + try: + c.call("POST", f"{base}/abort", ref) + except Exception: + pass # the dashboard's "unfinished uploads" cleanup catches it + raise + + +def upload_delivery( + c: Client, folder: Path, delivery_id: str, *, clear_stale=False, log=print +) -> dict: + folder = Path(folder) + files = sorted(p for p in folder.rglob("*") if p.is_file()) + if not files: + raise UploadError(f"{folder} has no files") + base = "/api/deliveries/" + urllib.parse.quote(delivery_id, safe="") + now = c.call("GET", base) + if now["uploads"]: + if not clear_stale: + raise UploadError( + f"{delivery_id} has {len(now['uploads'])} unfinished upload(s) (a " + "crashed or running upload); pass --clear-stale to drop them" + ) + for u in now["uploads"]: + c.call("POST", f"{base}/clear", {"record": u["record"]}) + total = sum(p.stat().st_size for p in files) or 1 + done = [0] + + def progress(n): + done[0] += n + log(f"\r{done[0] / total:6.1%}", end="", flush=True) + + names = [p.relative_to(folder).as_posix() for p in files] + for p, name in zip(files, names): + log(f"\r{name} ({p.stat().st_size / 1e6:.1f} MB)") + upload_file(c, base, p, name, progress) + log("\rmarking complete") + return c.call("POST", f"{base}/complete", {"files": names}) + + +def url_problem(url: str) -> str | None: + """Why the password must not be sent to `url`, or None. https only, except plain + http to this machine (localhost or a loopback address, exactly); never with + credentials inside the URL.""" + import ipaddress + + try: + u = urllib.parse.urlsplit(url) + host = u.hostname or "" + u.port # noqa: B018 (raises on a malformed port) + except ValueError: + return f"{url!r} is not a URL" + if u.username is not None or u.password is not None: + return "the URL must not contain a user or password; use --user" + if not host: + return f"{url!r} has no host" + if u.scheme == "https": + return None + if u.scheme == "http": + try: + loopback = ipaddress.ip_address(host).is_loopback + except ValueError: + loopback = host == "localhost" + if loopback: + return None + return "the dashboard URL must be https:// (plain http only to localhost)" + + +def main(argv: list[str] | None = None) -> int: + ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0]) + ap.add_argument("folder", type=Path) + ap.add_argument("--url", required=True, help="the dashboard, e.g. https://...") + ap.add_argument("--user", required=True) + ap.add_argument("--delivery-id", help="default: the folder's name") + ap.add_argument( + "--clear-stale", action="store_true", help="drop unfinished earlier uploads" + ) + args = ap.parse_args(argv) + problem = url_problem(args.url) + if problem: + print(f"error: {problem}", file=sys.stderr) + return 2 + password = os.environ.get("EGOVERSE_UPLOAD_PASSWORD") or getpass.getpass( + f"password for {args.user}: " + ) + c = Client(args.url) + try: + c.login(args.user, password) + out = upload_delivery( + c, + args.folder, + args.delivery_id or args.folder.resolve().name, + clear_stale=args.clear_stale, + ) + except (UploadError, OSError) as e: + print(f"\nerror: {e}", file=sys.stderr) + return 1 + print(f"\ndone: {len(out['files'])} file(s); the dashboard shows its progress") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/pyproject.toml b/pyproject.toml index acc7877a0..6cc81a7ba 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -84,6 +84,10 @@ dependencies = [ # not part of this package; install it by hand when running that converter. # The Ray data-processing image keeps its own requirements-ray.txt. +[project.optional-dependencies] +# The ingestion dashboard (egomimic/ingest/dashboard), deployed on its own. +dashboard = ["flask>=3"] + [tool.setuptools.packages.find] where = ["."] include = ["egomimic*"] diff --git a/tests/unit/test_ingest_dashboard.py b/tests/unit/test_ingest_dashboard.py new file mode 100644 index 000000000..981c685bc --- /dev/null +++ b/tests/unit/test_ingest_dashboard.py @@ -0,0 +1,361 @@ +"""The dashboard (egomimic/ingest/dashboard): accounts, login, and who sees what.""" + +import re + +import pytest + +pytest.importorskip("flask") + +from egomimic.ingest import accounts # noqa: E402 +from egomimic.ingest.dashboard import views # noqa: E402 +from egomimic.ingest.dashboard.app import create_app # noqa: E402 +from egomimic.ingest.store import LocalStore, Ref # noqa: E402 + +SECRET = "s" * 48 +PW = "correct horse battery" +A1 = Ref("acme", "episode_0001") +B1 = Ref("bolt", "episode_0001") + + +@pytest.fixture +def store(tmp_path, monkeypatch): + monkeypatch.setattr(accounts, "SCRYPT", {"n": 2**4, "r": 8, "p": 1}) # fast tests + s = LocalStore(tmp_path / "bucket") + accounts.add(s, "acme-ops", PW, role="vendor", vendor="acme") + accounts.add(s, "bolt-ops", PW, role="vendor", vendor="bolt") + accounts.add(s, "simar", PW, role="staff") + for ref in (A1, B1): + folder = tmp_path / f"src_{ref.vendor}" + folder.mkdir() + (folder / "episode.mcap").write_bytes(b"x" * 2048) + s.upload_delivery(ref, folder, {"by": "test"}) + return s + + +@pytest.fixture +def app(store): + return create_app(store, SECRET, secure_cookies=False) + + +def csrf(client, path="/login"): + page = client.get(path).get_data(as_text=True) + return re.search(r'name="csrf" value="([^"]+)"', page).group(1) + + +def login(app, user, password=PW): + c = app.test_client() + r = c.post("/login", data={"user": user, "password": password, "csrf": csrf(c)}) + return c, r + + +def ready(store, ref, **extra): + d = store.get_delivery(ref) + store.put_status( + ref, + { + "state": "ready", + "fingerprint": d.fingerprint, + "stage": "ab" * 32, + "report": {"num_frames": 3661, "valid_frac": {"left": 0.42}}, + "problems": [], + "history": [{"state": "ready", "at": "20261021T140312.000000Z"}], + **extra, + }, + ) + + +# ---------------------------------------------------------------- accounts + + +def test_passwords_are_stored_hashed_and_checked(store): + acct = store.get_accounts()["acme-ops"] + assert PW not in str(acct) and acct["role"] == "vendor" and acct["vendor"] == "acme" + assert accounts.check_password(acct, PW) + assert not accounts.check_password(acct, PW + "!") + assert not accounts.check_password(None, PW) # unknown user + + +def test_accounts_are_validated(store): + with pytest.raises(ValueError, match="already exists"): + accounts.add(store, "simar", PW, role="staff") + with pytest.raises(ValueError, match="at least"): + accounts.add(store, "new", "short", role="staff") + with pytest.raises(ValueError, match="vendor slug"): + accounts.add(store, "new", PW, role="vendor") + with pytest.raises(ValueError, match="no vendor"): + accounts.add(store, "new", PW, role="staff", vendor="acme") + with pytest.raises(ValueError): + accounts.add(store, "Bad.Name", PW, role="staff") + + +def test_the_cli_lists_accounts_without_hashes(store, capsys): + root = str(store.root) + assert accounts.main(["--local-store", root, "list"]) == 0 + out = capsys.readouterr().out + assert '"acme-ops"' in out and "scrypt" not in out + assert accounts.main(["--local-store", root, "remove", "bolt-ops"]) == 0 + assert "bolt-ops" not in store.get_accounts() + assert accounts.main(["--local-store", root, "remove", "nobody"]) == 1 + + +# ---------------------------------------------------------------- login + + +def test_pages_need_a_login(app): + c = app.test_client() + r = c.get("/") + assert r.status_code == 302 and "/login" in r.headers["Location"] + assert c.get("/d/acme/episode_0001").status_code == 302 + + +def test_a_wrong_password_or_unknown_user_is_refused(app): + for user, pw in (("acme-ops", "wrong password!"), ("nobody", PW)): + c, r = login(app, user, pw) + assert r.status_code == 401 and "Wrong user or password" in r.get_data(True) + assert c.get("/").status_code == 302 + + +def test_a_post_without_the_form_token_is_refused(app): + c = app.test_client() + c.get("/login") + r = c.post("/login", data={"user": "simar", "password": PW}) + assert r.status_code == 400 + + +def test_login_only_redirects_to_our_own_pages(app): + c = app.test_client() + for target, expect in (("/d/acme/episode_0001", "/d/acme/episode_0001"), + ("//evil.example", "/"), ("https://evil.example", "/")): # fmt: skip + r = c.post( + f"/login?next={target}", + data={"user": "simar", "password": PW, "csrf": csrf(c)}, # new per login + ) + assert r.headers["Location"] == expect + + +def test_logout_ends_the_session(app): + c, _ = login(app, "simar") + assert c.get("/").status_code == 200 + c.post("/logout", data={"csrf": csrf(c, "/")}) + assert c.get("/").status_code == 302 + + +def test_a_password_change_or_removal_ends_existing_sessions(app, store, monkeypatch): + from egomimic.ingest.dashboard import app as app_module + + monkeypatch.setattr(app_module, "ACCOUNTS_TTL", -1) # no caching + c, _ = login(app, "acme-ops") + assert c.get("/").status_code == 200 + accounts.set_password(store, "acme-ops", PW + " new") + assert c.get("/").status_code == 302 + c, _ = login(app, "bolt-ops") + accounts.remove(store, "bolt-ops") + assert c.get("/").status_code == 302 + + +def test_cookies_and_headers(store): + app = create_app(store, SECRET) # secure cookies by default + c, r = login(app, "simar") + cookie = r.headers["Set-Cookie"] + assert "Secure" in cookie and "HttpOnly" in cookie and "SameSite=Lax" in cookie + r = c.get("/login") + assert "frame-ancestors 'none'" in r.headers["Content-Security-Policy"] + assert r.headers["X-Content-Type-Options"] == "nosniff" + with pytest.raises(ValueError): + create_app(store, "too short") + + +# ---------------------------------------------------------------- who sees what + + +def test_a_vendor_sees_only_its_own_deliveries(app, store): + ready(store, A1) + ready(store, B1) + c, _ = login(app, "acme-ops") + page = c.get("/").get_data(as_text=True) + assert "/d/acme/episode_0001" in page and "/d/bolt/" not in page + assert c.get("/?vendor=bolt").get_data(as_text=True).count("/d/bolt/") == 0 + assert c.get("/d/acme/episode_0001").status_code == 200 + assert c.get("/d/bolt/episode_0001").status_code == 404 # as if it did not exist + assert c.get("/d/acme/no_such").status_code == 404 + + +def test_staff_see_every_vendor_and_can_filter(app, store): + ready(store, A1) + c, _ = login(app, "simar") + page = c.get("/").get_data(as_text=True) + assert "/d/acme/episode_0001" in page and "/d/bolt/episode_0001" in page + page = c.get("/?vendor=bolt").get_data(as_text=True) + assert "/d/bolt/" in page and "/d/acme/" not in page + page = c.get("/?state=ready").get_data(as_text=True) + assert "/d/acme/" in page and "/d/bolt/" not in page # bolt: no status yet + assert c.get("/d/bolt/episode_0001").status_code == 200 + assert c.get("/?vendor=../x").status_code == 404 + + +def test_architecture_is_not_part_of_dashboard(app): + staff, _ = login(app, "simar") + for path in ("/architecture", "/architecture/download"): + assert staff.get(path).status_code == 404 + assert 'href="/architecture"' not in staff.get("/").get_data(as_text=True) + + +def test_summary_respects_vendor_scope_but_not_state_filter(app, store): + ready(store, A1) + staff, _ = login(app, "simar") + vendor, _ = login(app, "bolt-ops") + + def metrics(client, path): + page = client.get(path).get_data(as_text=True) + return re.findall(r"(\d+)\s*([^<]+)", page) + + assert ("2", "Total deliveries") in metrics(staff, "/?state=ready") + assert ("1", "Awaiting review") in metrics(staff, "/") + assert ("1", "Total deliveries") in metrics(staff, "/?vendor=bolt") + assert ("0", "Awaiting review") in metrics(vendor, "/?vendor=acme") + + +def test_logins_are_throttled_before_any_password_work(store, monkeypatch): + from egomimic.ingest.dashboard import throttle as th + + now = [0.0] + app = create_app( + store, SECRET, secure_cookies=False, throttle=th.Throttle(lambda: now[0]) + ) + checks = [] + real = accounts.check_password + monkeypatch.setattr( + accounts, "check_password", lambda a, p: checks.append(1) or real(a, p) + ) + for _ in range(th.PAIR_FAILURES): + assert login(app, "acme-ops", "wrong password!")[1].status_code == 401 + c, r = login(app, "acme-ops") # even the right password waits now + assert r.status_code == 429 and int(r.headers["Retry-After"]) > 0 + assert "Too many attempts" in r.get_data(True) + assert len(checks) == th.PAIR_FAILURES # refused without hashing + assert login(app, "bolt-ops")[1].status_code == 302 # other accounts unaffected + now[0] += th.WINDOW # temporary: the window passes, no lasting lockout + assert login(app, "acme-ops")[1].status_code == 302 + + +def test_one_address_trying_many_accounts_is_throttled(): + from egomimic.ingest.dashboard import throttle as th + + t = th.Throttle(lambda: 0.0) + for i in range(th.IP_ATTEMPTS): + assert t.retry_after("1.2.3.4", f"user{i}") == 0 + assert t.retry_after("1.2.3.4", "another") > 0 + assert t.retry_after("5.6.7.8", "another") == 0 + + +def test_one_account_guessed_from_many_addresses_is_capped(): + from egomimic.ingest.dashboard import throttle as th + + t = th.Throttle(lambda: 0.0) + for i in range(th.USER_FAILURES): + assert t.retry_after(f"10.0.0.{i}", "acme-ops") == 0 + t.failed(f"10.0.0.{i}", "acme-ops") + assert t.retry_after("10.0.1.1", "acme-ops") > 0 + + +def test_behind_a_trusted_proxy_the_client_address_is_used(store): + from egomimic.ingest.dashboard import throttle as th + + app = create_app( + store, SECRET, secure_cookies=False, proxies=1, + throttle=th.Throttle(lambda: 0.0), + ) # fmt: skip + c = app.test_client() + for i in range(th.PAIR_FAILURES): + r = c.post( + "/login", + data={"user": "acme-ops", "password": "nope nope nope", "csrf": csrf(c)}, + headers={"X-Forwarded-For": "9.9.9.9"}, + ) + r = c.post( + "/login", + data={"user": "acme-ops", "password": PW, "csrf": csrf(c)}, + headers={"X-Forwarded-For": "8.8.8.8"}, # another client: not throttled + ) + assert r.status_code == 302 + + +def test_updated_is_the_last_status_write(store): + ready(store, A1) # history says 2026-10-21 14:03; put_status stamps updated_at + r = views.row(store, A1) + assert r.updated == r.status["updated_at"] != "20261021T140312.000000Z" + + +def test_the_detail_page_shows_the_conversion(app, store): + ready(store, A1, diff={"action": "new", "frames": 3661}) + c, _ = login(app, "acme-ops") + page = c.get("/d/acme/episode_0001").get_data(as_text=True) + assert "Converted; waiting for EgoVerse review" in page + assert "3661" in page and "left: 0.42" in page and "episode.mcap" in page + assert "2.0 KB" in page and "2026-10-21 14:03" in page # the history entry + + +def test_vendor_text_is_escaped(app, store): + ready(store, A1, state="rejected", problems=[""]) + c, _ = login(app, "acme-ops") + page = c.get("/d/acme/episode_0001").get_data(as_text=True) + assert "" not in page + assert "<script>" in page + + +def test_vendors_do_not_see_our_internals(app, store): + trace = "RuntimeError: disk full\nTraceback (most recent call last):\n File ..." + ready(store, A1, state="failed", error=trace) + ready( + store, B1, state="failed", + error="ValueError: unsupported lens in /coc/flash7/x/work/b/episode.mcap", + decision={"decision": "decline", "who": "simar", "comment": "blurry"}, + ) # fmt: skip + c, _ = login(app, "acme-ops") + page = c.get("/d/acme/episode_0001").get_data(as_text=True) + assert "disk full" not in page and "Traceback" not in page + assert "Failed on our side" in page + c, _ = login(app, "bolt-ops") + page = c.get("/d/bolt/episode_0001").get_data(as_text=True) + assert "unsupported lens in <path>" in page and "/coc/" not in page + assert "blurry" in page and "simar" not in page # the decision, not who made it + c, _ = login(app, "simar") + page = c.get("/d/acme/episode_0001").get_data(as_text=True) + assert "Traceback" in page # staff see everything + + +def test_states_follow_the_bucket_when_it_moves_on(store, tmp_path): + ready(store, A1) + assert views.row(store, A1).state == "ready" + store.start_upload(A1, "episode.mcap") # a correction begins + assert views.row(store, A1).state == "uploading" + assert views.state_of(None, None) == "queued" + listed = store.get_delivery(B1) + assert views.state_of(listed, {"state": "ready", "fingerprint": "old"}) == "queued" + + +def test_staff_opening_the_upload_page_are_sent_back_with_a_reason(app): + c, _ = login(app, "simar") + r = c.get("/new") + assert r.status_code == 302 and r.headers["Location"] == "/" + assert "Uploads are made from vendor accounts." in c.get("/").get_data(True) + + +def test_errors_use_the_dashboard_layout_and_the_api_keeps_json(app): + c, _ = login(app, "acme-ops") + r = c.get("/d/bolt/episode_0001") + page = r.get_data(as_text=True) + assert r.status_code == 404 and "Not Found" in page and "Back to" in page + assert 'href="/style.css"' in page + r = c.get("/api/deliveries/x/nonexistent") + assert r.status_code == 404 and "error" in r.get_json() + r = c.post("/api/deliveries/x/uploads", json={}, headers={"X-CSRF-Token": "bad"}) + assert r.status_code == 400 and r.get_json()["error"] + + +def test_nested_values_read_as_text(): + from egomimic.ingest.dashboard.templates import fmt + + assert fmt({"head": 0.9918055, "left": 0.4176}) == "head: 0.992, left: 0.418" + assert fmt({"new": {"head": 0.5}}) == "new: head: 0.5" + assert fmt([]) == "none" and fmt(None) == "none" and fmt(True) == "true" diff --git a/tests/unit/test_ingest_review.py b/tests/unit/test_ingest_review.py new file mode 100644 index 000000000..767875130 --- /dev/null +++ b/tests/unit/test_ingest_review.py @@ -0,0 +1,296 @@ +"""Staff review in the dashboard (egomimic/ingest/dashboard/review.py): approve and +decline, one at a time or in bulk, only ever on the result the reviewer saw.""" + +import re +import shutil +import subprocess + +import pytest + +pytest.importorskip("flask") + +from egomimic.ingest import accounts # noqa: E402 +from egomimic.ingest.dashboard import views # noqa: E402 +from egomimic.ingest.dashboard.app import create_app # noqa: E402 +from egomimic.ingest.store import LocalStore, Ref # noqa: E402 + +SECRET = "s" * 48 +PW = "correct horse battery" +A = Ref("acme", "episode_0001") +B = Ref("acme", "episode_0002") +C = Ref("bolt", "episode_0001") + + +@pytest.fixture +def store(tmp_path, monkeypatch): + monkeypatch.setattr(accounts, "SCRYPT", {"n": 2**4, "r": 8, "p": 1}) + s = LocalStore(tmp_path / "bucket") + accounts.add(s, "acme-ops", PW, role="vendor", vendor="acme") + accounts.add(s, "simar", PW, role="staff") + accounts.add(s, "ryan-co", PW, role="staff") + return s + + +@pytest.fixture +def app(store): + return create_app(store, SECRET, secure_cookies=False) + + +def login(app, user): + c = app.test_client() + page = c.get("/login").get_data(as_text=True) + token = re.search(r'name="csrf" value="([^"]+)"', page).group(1) + c.post("/login", data={"user": user, "password": PW, "csrf": token}) + page = c.get("/").get_data(as_text=True) + c.csrf = re.search(r'name="csrf" value="([^"]+)"', page).group(1) + return c + + +def converted(store, tmp_path, ref, stage="ab" * 32, data=b"x"): + """A delivery the watcher has staged: complete, with a ready status.""" + folder = tmp_path / f"src_{ref.vendor}_{ref.delivery_id}_{data.hex()[:8]}" + folder.mkdir() + (folder / "episode.mcap").write_bytes(data) + store.upload_delivery(ref, folder, {"by": "test"}) + store.put_status( + ref, + { + "state": "ready", + "stage": stage, + "fingerprint": store.get_delivery(ref).fingerprint, + "report": {"num_frames": 60}, + }, + ) + + +def decide(c, ref, decision, stage="ab" * 32, comment=""): + return c.post( + f"/d/{ref.vendor}/{ref.delivery_id}/decide", + data={"decision": decision, "stage": stage, "comment": comment, "csrf": c.csrf}, + follow_redirects=True, + ) + + +def test_staff_approve_the_result_they_saw(app, store, tmp_path): + converted(store, tmp_path, A) + c = login(app, "simar") + page = c.get("/d/acme/episode_0001").get_data(as_text=True) + assert "Approve" in page and "Result abababababab" in page + page = decide(c, A, "approve").get_data(as_text=True) + assert "Approved; the next pipeline pass acts on it." in page + (d,) = store.decisions(A) + assert (d["decision"], d["who"], d["stage"]) == ("approve", "simar", "ab" * 32) + assert views.row(store, A).state == "approved" + assert "currently\napproved by simar" in page + vendor = login(app, "acme-ops") + page = vendor.get("/d/acme/episode_0001").get_data(as_text=True) + assert "Approved; added to the dataset at the next pass" in page + assert "simar" not in page and "

Review

" not in page + + +def test_a_decline_needs_a_comment_and_the_vendor_sees_it(app, store, tmp_path): + converted(store, tmp_path, A) + c = login(app, "simar") + page = decide(c, A, "decline").get_data(as_text=True) + assert "A decline needs a comment" in page and store.decisions(A) == [] + decide(c, A, "decline", comment="Camera faces the ceiling") + assert views.row(store, A).state == "declined" + page = login(app, "acme-ops").get("/d/acme/episode_0001").get_data(as_text=True) + assert "Declined by EgoVerse" in page and "Camera faces the ceiling" in page + assert "simar" not in page + + +def test_a_reviewer_can_change_their_mind_until_the_watcher_acts(app, store, tmp_path): + converted(store, tmp_path, A) + c = login(app, "simar") + decide(c, A, "approve") + decide(c, A, "decline", comment="wrong task") + assert views.row(store, A).state == "declined" + assert store.latest_decisions()[A]["decision"] == "decline" + + +def test_nothing_is_recorded_for_a_result_the_reviewer_did_not_see( + app, store, tmp_path +): + converted(store, tmp_path, A, stage="cd" * 32) # re-converted since the page loaded + c = login(app, "simar") + page = decide(c, A, "approve", stage="ab" * 32).get_data(as_text=True) + assert "Not recorded" in page and store.decisions(A) == [] + + converted(store, tmp_path, B) + folder = tmp_path / "v2" + folder.mkdir() + (folder / "episode.mcap").write_bytes(b"v2") + store.upload_delivery(B, folder, {}) # re-uploaded after conversion + page = decide(c, B, "approve").get_data(as_text=True) + assert "re-uploaded" in page and store.decisions(B) == [] + + store.start_upload(B, "episode.mcap") # a correction under way + page = decide(c, B, "approve").get_data(as_text=True) + assert "upload to it is under way" in page and store.decisions(B) == [] + + +def test_only_ready_deliveries_can_be_decided(app, store, tmp_path): + converted(store, tmp_path, A) + store.put_status(A, {**store.get_status(A), "state": "committed"}) + c = login(app, "simar") + assert "

Review

" not in c.get("/d/acme/episode_0001").get_data(True) + assert "Not recorded" in decide(c, A, "approve").get_data(True) + + +def test_vendors_cannot_decide(app, store, tmp_path): + converted(store, tmp_path, A) + c = login(app, "acme-ops") + r = c.post( + "/d/acme/episode_0001/decide", + data={"decision": "approve", "stage": "ab" * 32, "csrf": c.csrf}, + ) + assert r.status_code == 404 + r = c.post( + "/decide", + data={"decision": "approve", "item": f"acme/episode_0001/{'ab' * 32}", + "csrf": c.csrf}, + ) # fmt: skip + assert r.status_code == 404 and store.decisions(A) == [] + + +def test_decisions_need_the_form_token(app, store, tmp_path): + converted(store, tmp_path, A) + c = login(app, "simar") + c.csrf = "forged" + assert decide(c, A, "approve").status_code == 400 + assert store.decisions(A) == [] + + +def test_bulk_approval_decides_each_and_skips_what_changed(app, store, tmp_path): + converted(store, tmp_path, A) + converted(store, tmp_path, B) + converted(store, tmp_path, C, stage="ef" * 32) + store.put_status(B, {**store.get_status(B), "stage": "99" * 32}) # re-converted + c = login(app, "simar") + page = c.get("/").get_data(as_text=True) + assert page.count('class="pick"') == 3 and 'id="all"' in page + r = c.post( + "/decide", + data={ + "decision": "approve", + "item": [f"acme/episode_0001/{'ab' * 32}", f"acme/episode_0002/{'ab' * 32}", + f"bolt/episode_0001/{'ef' * 32}"], + "back": "/?state=ready", + "csrf": c.csrf, + }, + ) # fmt: skip + assert r.headers["Location"] == "/?state=ready" + page = c.get("/").get_data(as_text=True) + assert "Approved 2; the next pipeline pass acts on them." in page + assert "Skipped 1" in page and "acme/episode_0002" in page + assert [len(store.decisions(r)) for r in (A, B, C)] == [1, 0, 1] + assert store.decisions(C)[0]["stage"] == "ef" * 32 # each its own stage + + +def test_bulk_decisions_are_checked(app, store, tmp_path): + converted(store, tmp_path, A) + c = login(app, "simar") + item = f"acme/episode_0001/{'ab' * 32}" + post = lambda **d: c.post("/decide", data={"csrf": c.csrf, **d}) # noqa: E731 + assert post(decision="approve", item=["acme/../x/" + "ab" * 32]).status_code == 400 + assert post(decision="approve", item=[item] * 501).status_code == 400 + assert post(decision="maybe", item=[item]).status_code == 400 + post(decision="decline", item=[item]) # no comment + post(decision="approve") # nothing ticked + page = c.get("/").get_data(as_text=True) + assert "A decline needs a comment" in page and "Tick the deliveries" in page + assert store.decisions(A) == [] + r = post(decision="approve", item=[item], back="//evil.example") + assert r.headers["Location"] == "/" + + +def test_latest_decisions_come_from_one_listing(store): + store.add_decision(A, "approve", "ryan-co", stage="ab" * 32) + store.add_decision(A, "decline", "simar", "blurry", stage="ab" * 32) + store.add_decision(C, "approve", "simar", stage="cd" * 32) + store._put("pipeline/decisions/acme/episode_0001/garbage.json", b"{}") + latest = store.latest_decisions() + assert latest[A]["decision"] == "decline" and latest[A]["who"] == "simar" + assert latest[C] == {**latest[C], "who": "simar", "stage": "cd" * 32} + assert list(store.latest_decisions("bolt")) == [C] + assert store.decisions(A)[-1]["comment"] == "blurry" # the record keeps it + + +# ---------------------------------------------------------------- the whole loop + +needs_ffmpeg = pytest.mark.skipif( + not ( + shutil.which("ffmpeg") + and "libx264" + in subprocess.run( + ["ffmpeg", "-hide_banner", "-encoders"], capture_output=True + ).stdout.decode() + ), + reason="needs ffmpeg with libx264", +) + + +@needs_ffmpeg +def test_upload_convert_approve_commit(app, store, tmp_path): + """A vendor's delivery, staged by the watcher, approved in the dashboard, committed + by the next pass.""" + lance = pytest.importorskip("lance") + from fixtures.mcap_episodes import write_mcap_episode + + from egomimic.ingest.launch import LocalLauncher + from egomimic.ingest.run import commit_round + from egomimic.ingest.watch import run_pass + + root, work = tmp_path / "lance", tmp_path / "work" + staging = str(tmp_path / "lance_staging") + launcher = LocalLauncher( + store, str(root), tmp_path / "jobs", codec="libx264", mano=False + ) + + def watch(): + return run_pass( + store, launcher, work, raw_root=str(tmp_path / "raw"), + lance_staging=staging, revision="r1", + ) # fmt: skip + + store.upload_delivery(A, write_mcap_episode(tmp_path / "ep"), {"by": "acme-ops"}) + assert watch().staged == ["acme/episode_0001"] + c = login(app, "simar") + page = c.get("/d/acme/episode_0001").get_data(as_text=True) + stage = re.search(r'name="stage" value="([0-9a-f]{64})"', page).group(1) + decide(c, A, "approve", stage=stage) + assert watch().approved == ["acme/episode_0001"] + commit_round(staging, str(root), work) + assert watch().committed == ["acme/episode_0001"] + assert lance.dataset(root / "recordings.lance").count_rows() == 1 + page = login(app, "acme-ops").get("/d/acme/episode_0001").get_data(as_text=True) + assert "In the dataset" in page + + +def test_a_decline_can_become_an_approval_before_the_watcher_acts(app, store, tmp_path): + converted(store, tmp_path, A) + c = login(app, "simar") + decide(c, A, "decline", comment="looked blurry") + assert views.row(store, A).state == "declined" + page = c.get("/d/acme/episode_0001").get_data(as_text=True) + assert "

Review

" in page # still reviewable + assert 'class="pick"' in c.get("/").get_data(as_text=True) # and in bulk + decide(c, A, "approve") + assert views.row(store, A).state == "approved" + assert store.latest_decisions()[A]["decision"] == "approve" + # Once the watcher has acted, it is no longer up for review. + store.put_status(A, {**store.get_status(A), "state": "declined"}) + assert "

Review

" not in c.get("/d/acme/episode_0001").get_data(True) + + +def test_a_re_uploaded_delivery_is_not_offered_for_review(app, store, tmp_path): + converted(store, tmp_path, A) + folder = tmp_path / "v2" + folder.mkdir() + (folder / "episode.mcap").write_bytes(b"v2") + store.upload_delivery(A, folder, {}) + c = login(app, "simar") + assert views.row(store, A).state == "queued" + assert "

Review

" not in c.get("/d/acme/episode_0001").get_data(True) + assert 'class="pick"' not in c.get("/").get_data(as_text=True) diff --git a/tests/unit/test_ingest_upload.py b/tests/unit/test_ingest_upload.py new file mode 100644 index 000000000..fcfd8ff54 --- /dev/null +++ b/tests/unit/test_ingest_upload.py @@ -0,0 +1,438 @@ +"""Dashboard uploads (egomimic/ingest/dashboard/uploads.py) and the command-line +uploader (egomimic/ingest/upload.py), on the local-directory store.""" + +import io +import json +import types +import urllib.error + +import pytest + +pytest.importorskip("flask") + +from egomimic.ingest import accounts, upload # noqa: E402 +from egomimic.ingest.dashboard import uploads # noqa: E402 +from egomimic.ingest.dashboard.app import create_app # noqa: E402 +from egomimic.ingest.store import LocalStore, R2Store, Ref # noqa: E402 + +SECRET = "s" * 48 +PW = "correct horse battery" +A = Ref("acme", "episode_0001") + + +@pytest.fixture +def store(tmp_path, monkeypatch): + monkeypatch.setattr(accounts, "SCRYPT", {"n": 2**4, "r": 8, "p": 1}) + s = LocalStore(tmp_path / "bucket") + accounts.add(s, "acme-ops", PW, role="vendor", vendor="acme") + accounts.add(s, "bolt-ops", PW, role="vendor", vendor="bolt") + accounts.add(s, "simar", PW, role="staff") + return s + + +@pytest.fixture +def app(store): + return create_app(store, SECRET, secure_cookies=False) + + +class Api: + """A logged-in JSON client over Flask's test client.""" + + def __init__(self, app, user): + self.c = app.test_client() + r = self.c.post("/api/login", json={"user": user, "password": PW}) + assert r.status_code == 200, r.get_json() + self.csrf = r.get_json()["csrf"] + + def __call__(self, method, path, body=None, **kw): + headers = {"X-CSRF-Token": self.csrf} + r = self.c.open(path, method=method, json=body, headers=headers, **kw) + return r.status_code, r.get_json() + + +def send(api, delivery, name, data: bytes): + """Upload one file through the API (local store: parts via /part).""" + base = f"/api/deliveries/{delivery}" + code, start = api("POST", f"{base}/uploads", {"name": name, "size": len(data)}) + assert code == 200, start + ref = {"name": name, "upload_id": start["upload_id"]} + ps = start["part_size"] + parts = list(range(1, start["parts"] + 1)) + code, out = api("POST", f"{base}/urls", {**ref, "parts": parts}) + assert code == 200 and all(u is None for u in out["urls"].values()) + etags = [] + for n in parts: + r = api.c.put( + f"{base}/part", + query_string={**ref, "part": n}, + data=data[(n - 1) * ps : n * ps], + headers={"X-CSRF-Token": api.csrf}, + ) + assert r.status_code == 200, r.get_json() + etags.append([n, r.get_json()["etag"]]) + return api("POST", f"{base}/finish", {**ref, "parts": etags}) + + +def test_a_vendor_uploads_a_delivery_through_the_api(app, store, monkeypatch): + monkeypatch.setattr(uploads, "MIN_PART", 4) # several parts from a small file + api = Api(app, "acme-ops") + assert send(api, "episode_0001", "episode.mcap", b"0123456789")[0] == 200 + assert send(api, "episode_0001", "annotation.mcap", b"labels")[0] == 200 + d = store.get_delivery(A) + assert not d.complete # not until the vendor says the delivery is all there + code, out = api( + "POST", "/api/deliveries/episode_0001/complete", + {"files": ["episode.mcap", "annotation.mcap"]}, + ) # fmt: skip + assert code == 200 and out["files"] == {"annotation.mcap": 6, "episode.mcap": 10} + d = store.get_delivery(A) + assert d.complete and store.completion(A)["by"] == "acme-ops" + path = store.root / "ingest-acme/episode_0001/episode.mcap" + assert path.read_bytes() == b"0123456789" + code, out = api("GET", "/api/deliveries/episode_0001") + assert out["state"] == "queued" and out["uploads"] == [] + + +def test_uploads_go_only_to_the_vendors_own_folder(app, store): + api = Api(app, "acme-ops") + send(api, "episode_0001", "episode.mcap", b"x") + assert store.get_delivery(Ref("acme", "episode_0001")) + assert store.get_delivery(Ref("bolt", "episode_0001")) is None + # Another vendor cannot use acme's upload id, even for the same delivery name. + code, start = api( + "POST", "/api/deliveries/episode_0002/uploads", + {"name": "episode.mcap", "size": 1}, + ) # fmt: skip + bolt = Api(app, "bolt-ops") + ref = {"name": "episode.mcap", "upload_id": start["upload_id"]} + assert ( + bolt("POST", "/api/deliveries/episode_0002/urls", {**ref, "parts": [1]})[0] + == 404 + ) + assert bolt("POST", "/api/deliveries/episode_0002/abort", ref)[0] == 404 + # ... nor can acme use it for another delivery or file. + assert api("POST", "/api/deliveries/other/urls", {**ref, "parts": [1]})[0] == 404 + wrong = {**ref, "name": "annotation.mcap"} + assert ( + api("POST", "/api/deliveries/episode_0002/finish", {**wrong, "parts": []})[0] + == 404 + ) + + +def test_staff_and_strangers_do_not_upload(app): + staff = Api(app, "simar") + body = {"name": "episode.mcap", "size": 1} + assert staff("POST", "/api/deliveries/x/uploads", body)[0] == 403 + assert app.test_client().get("/api/deliveries/x").status_code == 401 + assert staff.c.get("/new").status_code == 302 # back to the list, with a reason + + +def test_api_writes_need_the_csrf_token(app): + api = Api(app, "acme-ops") + r = api.c.post( + "/api/deliveries/x/uploads", json={"name": "episode.mcap", "size": 1} + ) + assert r.status_code == 400 + r = api.c.post("/api/login", data={"user": "acme-ops", "password": PW}) + assert r.status_code == 415 # the API login takes JSON only (no cross-site forms) + + +def test_files_no_format_accepts_are_refused_before_uploading(app, monkeypatch): + api = Api(app, "acme-ops") + start = "/api/deliveries/episode_0001/uploads" + code, out = api("POST", start, {"name": "recording.vrs", "size": 1}) + assert code == 422 and "episode.mcap" in out["error"] + code, out = api("POST", start, {"name": "episode.mcap", "size": 300 * 1024**3}) + assert code == 422 and "GiB limit" in out["error"] + for bad in ("../x/episode.mcap", "/etc/episode.mcap", "_complete.json"): + assert api("POST", start, {"name": bad, "size": 1})[0] == 400 + assert api("POST", start, {"name": "episode.mcap", "size": -1})[0] == 400 + assert api("POST", "/api/deliveries/..%2Fbolt/uploads", {})[0] in (400, 404) + + +def test_completion_waits_for_uploads_and_needs_a_whole_delivery(app, store): + api = Api(app, "acme-ops") + base = "/api/deliveries/episode_0001" + send(api, "episode_0001", "annotation.mcap", b"labels") + code, out = api("POST", f"{base}/complete", {"files": ["annotation.mcap"]}) + assert code == 422 and out["formats"][0]["required"] == ["episode.mcap"] + code, out = api("POST", f"{base}/complete", {"files": ["episode.mcap"]}) + assert code == 422 # listed but never uploaded + code, start = api("POST", f"{base}/uploads", {"name": "episode.mcap", "size": 1}) + code, out = api("POST", f"{base}/complete", {"files": ["annotation.mcap"]}) + assert code in (409, 422) + code, out = api( + "POST", f"{base}/complete", {"files": ["episode.mcap", "annotation.mcap"]} + ) + assert code == 409 and len(out["uploads"]) == 1 # the tab that was closed + assert api("POST", f"{base}/clear", {"record": "f" * 64})[0] == 404 + assert api("POST", f"{base}/clear", {"record": out["uploads"][0]})[0] == 200 + assert api("GET", base)[1]["uploads"] == [] + + +def test_part_sizes_stay_within_r2_limits(): + assert uploads.part_size(0) == uploads.MIN_PART + assert uploads.part_size(10 * 1024**3) == uploads.MIN_PART + big = 200 * 1024**3 + ps = uploads.part_size(big) + assert ps % 1024**2 == 0 and -(-big // ps) <= uploads.MAX_PARTS + + +def test_presigned_stores_send_parts_to_the_bucket(store, monkeypatch): + class FakeMeta: + endpoint_url = "https://acct.r2.cloudflarestorage.com" + + fake = types.SimpleNamespace(meta=FakeMeta()) + r2 = R2Store("rldb", client=fake) + assert r2.upload_origins == [ + "https://acct.r2.cloudflarestorage.com", + "https://*.acct.r2.cloudflarestorage.com", + ] + monkeypatch.setattr(LocalStore, "upload_origins", r2.upload_origins) + monkeypatch.setattr(LocalStore, "part_url", lambda self, *a, **k: "https://p") + app = create_app(store, SECRET, secure_cookies=False) + api = Api(app, "acme-ops") + csp = api.c.get("/new").headers["Content-Security-Policy"] + assert "connect-src 'self' https://acct.r2.cloudflarestorage.com" in csp + base = "/api/deliveries/episode_0001" + code, start = api("POST", f"{base}/uploads", {"name": "episode.mcap", "size": 1}) + ref = {"name": "episode.mcap", "upload_id": start["upload_id"]} + assert api("POST", f"{base}/urls", {**ref, "parts": [1]})[1]["urls"] == { + "1": "https://p" + } + r = api.c.put( + f"{base}/part", query_string={**ref, "part": 1}, data=b"x", + headers={"X-CSRF-Token": api.csrf}, + ) # fmt: skip + assert r.status_code == 404 # bytes never go through the dashboard then + + +def test_the_upload_page_and_uploader_are_served(app): + api = Api(app, "acme-ops") + page = api.c.get("/new?delivery=episode_0001").get_data(as_text=True) + assert "Correct episode_0001" in page and "episode.mcap" in page + assert 'src="/upload.js"' in page + assert "presigned" in api.c.get("/upload.js").get_data(as_text=True) + r = api.c.get("/upload.py") + assert r.status_code == 200 and b"def upload_delivery" in r.data + assert "attachment" in r.headers["Content-Disposition"] + assert "New delivery" in api.c.get("/").get_data(as_text=True) + + +# ---------------------------------------------------------------- the CLI + + +class FakeOpener: + """urllib's opener, answered by Flask's test client (cookies kept by the client).""" + + def __init__(self, app): + self.c = app.test_client() + + def open(self, req, timeout=None): + url = req.full_url.removeprefix("http://dash") + r = self.c.open( + url, + method=req.get_method(), + data=req.data, + headers=dict(req.header_items()), + ) + if r.status_code >= 400: + raise urllib.error.HTTPError( + url, r.status_code, r.status, {}, io.BytesIO(r.data) + ) + return _Resp(r.data) + + +class _Resp(io.BytesIO): + def __enter__(self): + return self + + def __exit__(self, *a): + self.close() + + +def cli(app): + c = upload.Client("http://dash", opener=FakeOpener(app)) + c.login("acme-ops", PW) + return c + + +def test_the_cli_uploads_a_folder_and_a_correction_replaces_it( + app, store, tmp_path, monkeypatch +): + monkeypatch.setattr(uploads, "MIN_PART", 7) + folder = tmp_path / "episode_0001" + folder.mkdir() + (folder / "episode.mcap").write_bytes(bytes(range(256)) * 3) + (folder / "annotation.mcap").write_bytes(b"labels v1") + out = upload.upload_delivery( + cli(app), folder, "episode_0001", log=lambda *a, **k: 0 + ) + assert out["files"] == {"annotation.mcap": 9, "episode.mcap": 768} + d = store.get_delivery(A) + assert d.complete + bucket = store.root / "ingest-acme/episode_0001" + assert (bucket / "episode.mcap").read_bytes() == bytes(range(256)) * 3 + + (folder / "annotation.mcap").unlink() # the correction drops the labels + (folder / "episode.mcap").write_bytes(b"v2") + upload.upload_delivery(cli(app), folder, "episode_0001", log=lambda *a, **k: 0) + assert sorted(p.name for p in bucket.iterdir() if p.name != "_complete.json") == [ + "episode.mcap" + ] + assert store.get_delivery(A).complete + + +def test_the_cli_stops_on_unfinished_uploads_unless_told(app, store, tmp_path): + folder = tmp_path / "f" + folder.mkdir() + (folder / "episode.mcap").write_bytes(b"x") + store.start_upload(A, "episode.mcap") # a crashed earlier attempt + with pytest.raises(upload.UploadError, match="clear-stale"): + upload.upload_delivery(cli(app), folder, "episode_0001", log=lambda *a, **k: 0) + upload.upload_delivery( + cli(app), folder, "episode_0001", clear_stale=True, log=lambda *a, **k: 0 + ) + assert store.get_delivery(A).complete + + +def test_the_cli_aborts_a_file_that_fails(app, store, tmp_path, monkeypatch): + folder = tmp_path / "f" + folder.mkdir() + (folder / "episode.mcap").write_bytes(b"x" * 100) + monkeypatch.setattr(upload, "RETRIES", 1) + c = cli(app) + real = c.call + + def flaky(method, path, *a, **kw): + if path.endswith("/part"): + raise upload.UploadError("network down") + return real(method, path, *a, **kw) + + c.call = flaky + with pytest.raises(upload.UploadError, match="network down"): + upload.upload_delivery(c, folder, "episode_0001", log=lambda *a, **k: 0) + assert store.uploads_in_progress(A) == [] # aborted, nothing left behind + assert not (store.root / "ingest-acme/episode_0001/_complete.json").exists() + + +def test_the_cli_refuses_plain_http_to_a_remote_dashboard(capsys): + assert upload.main(["x", "--url", "http://dash.example", "--user", "u"]) == 2 + assert "https" in capsys.readouterr().err + + +def test_the_api_login_reports_errors_as_json(app): + r = app.test_client().post( + "/api/login", json={"user": "acme-ops", "password": "no"} + ) + assert r.status_code == 401 and json.loads(r.data)["error"].startswith("Wrong") + + +# ---------------------------------------------------------------- a modified client + + +def started(api, data_len, monkeypatch, min_part=4): + monkeypatch.setattr(uploads, "MIN_PART", min_part) + base = "/api/deliveries/episode_0001" + code, start = api( + "POST", f"{base}/uploads", {"name": "episode.mcap", "size": data_len} + ) + assert code == 200 and (start["part_size"], start["parts"]) == (4, 3) + return base, {"name": "episode.mcap", "upload_id": start["upload_id"]} + + +def put(api, base, ref, n, data): + r = api.c.put( + f"{base}/part", query_string={**ref, "part": n}, data=data, + headers={"X-CSRF-Token": api.csrf}, + ) # fmt: skip + return r.status_code, r.get_json() + + +def test_links_only_for_the_declared_parts(app, monkeypatch): + api = Api(app, "acme-ops") + base, ref = started(api, 10, monkeypatch) + assert api("POST", f"{base}/urls", {**ref, "parts": [1, 2, 3]})[0] == 200 + for extra in ([4], [0], [10_000], ["1"]): + assert api("POST", f"{base}/urls", {**ref, "parts": extra})[0] == 400 + + +def test_parts_must_have_their_exact_length(app, store, monkeypatch): + api = Api(app, "acme-ops") + base, ref = started(api, 10, monkeypatch) + assert put(api, base, ref, 1, b"0123" * 50)[0] == 413 # oversized: not even read + assert put(api, base, ref, 3, b"89xx")[0] == 400 # the last part is 2 bytes + assert put(api, base, ref, 4, b"0123")[0] == 400 # an extra part + assert put(api, base, ref, 1, b"0123")[0] == 200 + + +def test_finishing_needs_exactly_the_declared_parts(app, store, monkeypatch): + api = Api(app, "acme-ops") + base, ref = started(api, 10, monkeypatch) + etags = [ + [n, put(api, base, ref, n, chunk)[1]["etag"]] + for n, chunk in ((1, b"0123"), (2, b"4567"), (3, b"89")) + ] + code, out = api("POST", f"{base}/finish", {**ref, "parts": etags[:2]}) # missing + assert code == 422 and "exactly parts 1..3" in out["error"] + code, out = api("POST", f"{base}/finish", {**ref, "parts": etags + [[4, "x"]]}) + assert code == 422 # extra + assert api("POST", f"{base}/finish", {**ref, "parts": etags})[0] == 200 + + +def test_a_file_bigger_than_declared_is_removed(app, store, monkeypatch): + """A client that gets bytes past the part checks (on R2: a bucket not enforcing the + signed length) still cannot complete a file other than the size it declared.""" + api = Api(app, "acme-ops") + base, ref = started(api, 10, monkeypatch) + uid = ref["upload_id"] + etags = [ + [n, store.upload_part(A, "episode.mcap", uid, n, chunk)] # behind the API + for n, chunk in ((1, b"0123"), (2, b"4567" * 10_000), (3, b"89")) + ] + code, out = api("POST", f"{base}/finish", {**ref, "parts": etags}) + assert code == 422 and "removed" in out["error"] + assert not (store.root / "ingest-acme/episode_0001/episode.mcap").exists() + assert store.uploads_in_progress(A) == [] + + +def test_internal_uploads_cannot_be_driven_through_the_api(app, store): + """An upload started without a declared size (internal tools) is not the API's.""" + api = Api(app, "acme-ops") + uid = store.start_upload(A, "episode.mcap") + ref = {"name": "episode.mcap", "upload_id": uid} + assert ( + api("POST", "/api/deliveries/episode_0001/urls", {**ref, "parts": [1]})[0] + == 404 + ) + + +@pytest.mark.parametrize( + "url", + [ + "http://localhost.attacker.example", + "http://127.0.0.1.attacker.example", + "http://localhost@attacker.example", + "https://user:pw@dash.example", + "http://dash.example", + "ftp://dash.example", + "http://10.0.0.1", + "https://", + "http://localhost:notaport", + ], +) +def test_the_password_only_goes_to_https_or_this_machine(url, monkeypatch, capsys): + monkeypatch.setattr( + upload, "Client", lambda *a: pytest.fail(f"would have logged in to {url}") + ) + assert upload.url_problem(url) + assert upload.main(["x", "--url", url, "--user", "u"]) == 2 + + +@pytest.mark.parametrize( + "url", + ["https://dash.example", "https://dash.example:8443/", "http://localhost:8770", + "http://127.0.0.1:8765", "http://[::1]:8000"], +) # fmt: skip +def test_https_and_loopback_urls_are_allowed(url): + assert upload.url_problem(url) is None diff --git a/uv.lock b/uv.lock index 4adf54eb7..d921779f8 100644 --- a/uv.lock +++ b/uv.lock @@ -426,6 +426,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/5a/9f/e4d4ff45bc63d22fa63c9fc3835c480e3ec6b71009d6338cb603394ef540/blessed-1.39.0-py3-none-any.whl", hash = "sha256:666e7e3fd0a4e38c3a262eaaf1e22a4ce2c81337aa17593c3f60ea136ec24fe1", size = 124254, upload-time = "2026-05-04T17:49:59.976Z" }, ] +[[package]] +name = "blinker" +version = "1.9.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/21/28/9b3f50ce0e048515135495f198351908d99540d69bfdc8c1d15b73dc55ce/blinker-1.9.0.tar.gz", hash = "sha256:b4ce2265a7abece45e7cc896e98dbebe6cead56bcf805a3d23136d145f5445bf", size = 22460, upload-time = "2024-11-08T17:25:47.436Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/10/cb/f2ad4230dc2eb1a74edf38f1a38b9b52277f75bef262d8908e60d957e13c/blinker-1.9.0-py3-none-any.whl", hash = "sha256:ba0efaa9080b619ff2f3459d1d500c57bddea4a6b424b60a91141db6fd2f08bc", size = 8458, upload-time = "2024-11-08T17:25:46.184Z" }, +] + [[package]] name = "boto3" version = "1.43.6" @@ -1085,6 +1094,11 @@ dependencies = [ { name = "zarr" }, ] +[package.optional-dependencies] +dashboard = [ + { name = "flask" }, +] + [package.metadata] requires-dist = [ { name = "argparse" }, @@ -1096,6 +1110,7 @@ requires-dist = [ { name = "datasets", specifier = "==4.0.0" }, { name = "dm-control", specifier = "==1.0.8" }, { name = "einops" }, + { name = "flask", marker = "extra == 'dashboard'", specifier = ">=3" }, { name = "foxglove-schemas-protobuf", specifier = "==0.4.0" }, { name = "geomloss" }, { name = "gpustat" }, @@ -1149,6 +1164,7 @@ requires-dist = [ { name = "wandb" }, { name = "zarr", specifier = "==3.1.5" }, ] +provides-extras = ["dashboard"] [[package]] name = "einops" @@ -1202,6 +1218,23 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/81/47/dd9a212ef6e343a6857485ffe25bba537304f1913bdbed446a23f7f592e1/filelock-3.29.0-py3-none-any.whl", hash = "sha256:96f5f6344709aa1572bbf631c640e4ebeeb519e08da902c39a001882f30ac258", size = 39812, upload-time = "2026-04-19T15:39:08.752Z" }, ] +[[package]] +name = "flask" +version = "3.1.3" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "blinker" }, + { name = "click" }, + { name = "itsdangerous" }, + { name = "jinja2" }, + { name = "markupsafe" }, + { name = "werkzeug" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/26/00/35d85dcce6c57fdc871f3867d465d780f302a175ea360f62533f12b27e2b/flask-3.1.3.tar.gz", hash = "sha256:0ef0e52b8a9cd932855379197dd8f94047b359ca0a78695144304cb45f87c9eb", size = 759004, upload-time = "2026-02-19T05:00:57.678Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/7f/9c/34f6962f9b9e9c71f6e5ed806e0d0ff03c9d1b0b2340088a0cf4bce09b18/flask-3.1.3-py3-none-any.whl", hash = "sha256:f4bcbefc124291925f1a26446da31a5178f9483862233b23c0c96a20701f670c", size = 103424, upload-time = "2026-02-19T05:00:56.027Z" }, +] + [[package]] name = "fonttools" version = "4.62.1" @@ -1855,6 +1888,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/7b/55/e5326141505c5d5e34c5e0935d2908a74e4561eca44108fbfb9c13d2911a/isoduration-20.11.0-py3-none-any.whl", hash = "sha256:b2904c2a4228c3d44f409c8ae8e2370eb21a26f7ac2ec5446df141dde3452042", size = 11321, upload-time = "2020-11-01T10:59:58.02Z" }, ] +[[package]] +name = "itsdangerous" +version = "2.2.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/9c/cb/8ac0172223afbccb63986cc25049b154ecfb5e85932587206f42317be31d/itsdangerous-2.2.0.tar.gz", hash = "sha256:e0050c0b7da1eea53ffaf149c0cfbb5c6e2e2b69c4bef22c81fa6eb73e5f6173", size = 54410, upload-time = "2024-04-16T21:28:15.614Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/04/96/92447566d16df59b2a776c0fb82dbc4d9e07cd95062562af01e408583fc4/itsdangerous-2.2.0-py3-none-any.whl", hash = "sha256:c6242fc49e35958c8b15141343aa660db5fc54d4f13a1db01a3f5891b98700ef", size = 16234, upload-time = "2024-04-16T21:28:14.499Z" }, +] + [[package]] name = "jedi" version = "0.20.0" @@ -1884,7 +1926,7 @@ name = "jinxed" version = "2.0.0" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "ansicon" }, + { name = "ansicon", marker = "sys_platform == 'win32'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/cb/eb/8821ce6e7386e96355f2c6be83944925b4a0870572896fd33a4e61b8aa5a/jinxed-2.0.0.tar.gz", hash = "sha256:64b960b8f8d9966e1b2e7cb57ea2b1aa0a8d7e68045b96dc7ef58201e43a1209", size = 127118, upload-time = "2026-05-08T21:25:25.917Z" } wheels = [ @@ -3430,7 +3472,7 @@ name = "nvidia-cudnn-cu12" version = "9.5.1.17" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "nvidia-cublas-cu12" }, + { name = "nvidia-cublas-cu12", marker = "sys_platform != 'emscripten' and sys_platform != 'win32'" }, ] wheels = [ { url = "https://files.pythonhosted.org/packages/2a/78/4535c9c7f859a64781e43c969a3a7e84c54634e319a996d43ef32ce46f83/nvidia_cudnn_cu12-9.5.1.17-py3-none-manylinux_2_28_x86_64.whl", hash = "sha256:30ac3869f6db17d170e0e556dd6cc5eee02647abc31ca856634d5a40f82c15b2", size = 570988386, upload-time = "2024-10-25T19:54:26.39Z" }, @@ -3441,7 +3483,7 @@ name = "nvidia-cufft-cu12" version = "11.3.0.4" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "nvidia-nvjitlink-cu12" }, + { name = "nvidia-nvjitlink-cu12", marker = "sys_platform != 'emscripten' and sys_platform != 'win32'" }, ] wheels = [ { url = "https://files.pythonhosted.org/packages/8f/16/73727675941ab8e6ffd86ca3a4b7b47065edcca7a997920b831f8147c99d/nvidia_cufft_cu12-11.3.0.4-py3-none-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:ccba62eb9cef5559abd5e0d54ceed2d9934030f51163df018532142a8ec533e5", size = 200221632, upload-time = "2024-11-20T17:41:32.357Z" }, @@ -3470,9 +3512,9 @@ name = "nvidia-cusolver-cu12" version = "11.7.1.2" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "nvidia-cublas-cu12" }, - { name = "nvidia-cusparse-cu12" }, - { name = "nvidia-nvjitlink-cu12" }, + { name = "nvidia-cublas-cu12", marker = "sys_platform != 'emscripten' and sys_platform != 'win32'" }, + { name = "nvidia-cusparse-cu12", marker = "sys_platform != 'emscripten' and sys_platform != 'win32'" }, + { name = "nvidia-nvjitlink-cu12", marker = "sys_platform != 'emscripten' and sys_platform != 'win32'" }, ] wheels = [ { url = "https://files.pythonhosted.org/packages/f0/6e/c2cf12c9ff8b872e92b4a5740701e51ff17689c4d726fca91875b07f655d/nvidia_cusolver_cu12-11.7.1.2-py3-none-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:e9e49843a7707e42022babb9bcfa33c29857a93b88020c4e4434656a655b698c", size = 158229790, upload-time = "2024-11-20T17:43:43.211Z" }, @@ -3484,7 +3526,7 @@ name = "nvidia-cusparse-cu12" version = "12.5.4.2" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "nvidia-nvjitlink-cu12" }, + { name = "nvidia-nvjitlink-cu12", marker = "sys_platform != 'emscripten' and sys_platform != 'win32'" }, ] wheels = [ { url = "https://files.pythonhosted.org/packages/06/1e/b8b7c2f4099a37b96af5c9bb158632ea9e5d9d27d7391d7eb8fc45236674/nvidia_cusparse_cu12-12.5.4.2-py3-none-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:7556d9eca156e18184b94947ade0fba5bb47d69cec46bf8660fd2c71a4b48b73", size = 216561367, upload-time = "2024-11-20T17:44:54.824Z" }, @@ -3693,7 +3735,7 @@ name = "pexpect" version = "4.9.0" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "ptyprocess" }, + { name = "ptyprocess", marker = "sys_platform != 'emscripten' and sys_platform != 'win32'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/42/92/cc564bf6381ff43ce1f4d06852fc19a2f11d180f23dc32d9588bee2f149d/pexpect-4.9.0.tar.gz", hash = "sha256:ee7d41123f3c9911050ea2c2dac107568dc43b2d3b0c7557a33212c398ead30f", size = 166450, upload-time = "2023-11-25T09:07:26.339Z" } wheels = [ @@ -3828,7 +3870,7 @@ resolution-markers = [ "python_full_version < '3.12' and sys_platform != 'emscripten' and sys_platform != 'win32'", ] dependencies = [ - { name = "numpy" }, + { name = "numpy", marker = "python_full_version < '3.12'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/53/ae/677391c5e75858b5496dcdf4146952646c13a8b31c04090fa98c37a04198/positional_encodings-6.0.3.tar.gz", hash = "sha256:c144ad52e78f09d16effcbfabdf3e84050b56a37e342248476e18bbffd747b24", size = 6626, upload-time = "2024-01-04T18:17:44.133Z" } wheels = [ @@ -3837,7 +3879,7 @@ wheels = [ [package.optional-dependencies] pytorch = [ - { name = "torch" }, + { name = "torch", marker = "python_full_version < '3.12'" }, ] [[package]] @@ -3856,7 +3898,7 @@ resolution-markers = [ "python_full_version == '3.12.*' and sys_platform != 'emscripten' and sys_platform != 'win32'", ] dependencies = [ - { name = "numpy" }, + { name = "numpy", marker = "python_full_version >= '3.12'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/6d/04/93a78a33088fe9115602883a39f8395cdb5561388a24e65f57c11920c638/positional_encodings-6.0.4.tar.gz", hash = "sha256:435db9596a73759e7caa8f677ebf6d2ee4d39187112244e4d8b6d7b62bc119ad", size = 6814, upload-time = "2024-10-23T07:33:48.273Z" } wheels = [ @@ -3865,7 +3907,7 @@ wheels = [ [package.optional-dependencies] pytorch = [ - { name = "torch" }, + { name = "torch", marker = "python_full_version >= '3.12'" }, ] [[package]] @@ -5780,7 +5822,7 @@ name = "triton" version = "3.3.1" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "setuptools" }, + { name = "setuptools", marker = "sys_platform != 'emscripten' and sys_platform != 'win32'" }, ] wheels = [ { url = "https://files.pythonhosted.org/packages/21/2f/3e56ea7b58f80ff68899b1dbe810ff257c9d177d288c6b0f55bf2fe4eb50/triton-3.3.1-cp311-cp311-manylinux_2_27_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:b31e3aa26f8cb3cc5bf4e187bf737cbacf17311e1112b781d4a059353dfd731b", size = 155689937, upload-time = "2025-05-29T23:39:44.182Z" }, @@ -5933,6 +5975,18 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/34/db/b10e48aa8fff7407e67470363eac595018441cf32d5e1001567a7aeba5d2/websocket_client-1.9.0-py3-none-any.whl", hash = "sha256:af248a825037ef591efbf6ed20cc5faa03d3b47b9e5a2230a529eeee1c1fc3ef", size = 82616, upload-time = "2025-10-07T21:16:34.951Z" }, ] +[[package]] +name = "werkzeug" +version = "3.1.9" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "markupsafe" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/a4/34/4dd12fc8bb7d61c91467ec3efe415ffa7d5456f799954b40c5bbaeae470e/werkzeug-3.1.9.tar.gz", hash = "sha256:55ca7c70a75689be937aa27f8ff4b018f06ff4838fc73045560bf0f5a1291060", size = 940188, upload-time = "2026-09-27T18:33:41.637Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/a1/38/df03f564f43cec2684823f3cccae1a652ee7face1cbaa76fb223096e64d7/werkzeug-3.1.9-py3-none-any.whl", hash = "sha256:6392e50c78460ba618e5b21f08a71f59c99ce99cdc6cf6e3dd7e6ccca8754fab", size = 228700, upload-time = "2026-09-27T18:33:39.685Z" }, +] + [[package]] name = "widgetsnbextension" version = "4.0.15"