# Project export: BOB Forge This document was generated by HackStack to give an AI agent context about a hackathon project. Sections are labeled with their provenance; content marked as truncated was cut to keep this document small. ## Project metadata - Hackathon: OpenAI Build Week - Tagline: Local Codex diagnoses the incident. BOB Forge isolates and tests the patch, binds phone approval to the exact bytes, deploys, verifies, and rolls back locally. - Devpost: https://devpost.com/software/bob-forge - GitHub: https://github.com/italy2288-art/bob-forge - Video: https://www.youtube.com/embed/Oz4vsLRWP4I?enablejsapi=1&hl=en_US&rel=0&start=&version=3&wmode=transparent - Team: 1 GitHub contributor(s) — Denys (86 commits) ## Devpost submission (written by the team) ### Inspiration The moment that triggered BOB Forge was simple: an HTTP health check said the service was alive, but its WebSocket feed had already stopped after close code 1012. A coding agent can propose a convincing fix, but a proposal is not a release. Someone still has to isolate the change, prove what was tested, decide whether those exact bytes may deploy, measure recovery, and preserve a rollback record. I built BOB Forge to explore one question: can an AI repair a real incident without becoming the release authority? ### What it does BOB Forge is a local-first release control plane for a real, intentionally broken reconnect service. The full flow runs on a Windows PC and can be steered from a phone on the same Wi-Fi network: The operator submits the stalled-feed incident. Locally authenticated Codex and GPT-5.6 inspect an isolated per-task Git copy and return a structured diagnosis and bounded repair proposal. BOB Forge validates the proposal, permits only the registered source edit, and applies it outside the model process. Fixed compile and reconnect gates run inside hardened local Docker containers while the original source remains untouched. A human reviews the result on a paired phone. Approval is bound to five integrity digests: the plan, tested artifact, baseline, candidate, and deployment manifest. Only the approved frozen bytes reach the local Docker sandbox. BOB Forge measures recovery, rehearses an exact rollback, and exports a portable evidence pack with a semantic verdict. The demonstrated repair changes one file. The broken client receives two messages and stops; the tested candidate reconnects with bounded exponential backoff and reaches the required message count. Why it is different Most agent demos focus on whether a model can write a patch. BOB Forge focuses on everything that must be true after the patch looks plausible. Codex owns reasoning. Deterministic software owns file scope, patch validation, commands, tests, approval, deployment, health measurement, rollback, and evidence. There is no automatic switch to a paid API provider, and the default demo uses the user's existing local Codex authentication rather than an OpenAI API key. This separation is useful for solo developers and small teams that want agentic speed but do not have a dedicated SRE or release engineering function. It turns “the model says the fix works” into a reviewable chain of observed facts. How I built it The control plane is written in Python and persists tasks, versioned state transitions, events, artifacts, approvals, deployments, and recovery state in SQLite. Durable server-sent events replay after reload. A responsive HTML, CSS, and JavaScript HUD exposes the same authoritative state on desktop and mobile. The local Codex provider consumes structured JSONL. A bounded inspector selects context from an isolated Git copy, and the source plus Git metadata are hashed before and after the model turn. The patch engine enforces selected-file membership, expected hashes, exact replacements, path and size limits, secret checks, and Python AST restrictions. Docker test and deployment commands are fixed by the product, not authored by the model. Approval is versioned and fails closed if any bound digest changes. The exported schema-v2 Evidence Pack can be checked by a standalone verifier without Codex, Docker, an API key, or phone access. How I used Codex and GPT-5.6 During Build Week, Codex was my implementation and review partner for the independent BOB Forge control plane, reconnect fixture, safe patch boundary, Docker gates, phone workflow, failure handling, tests, and submission packaging. It accelerated repository inspection, implementation, debugging, security review, and repeated verification. I made the product and release-authority decisions: one honest vertical slice, local Windows execution, explicit human approval, no source merge or push, and no automatic paid-provider fallback. Inside the product, the local Codex CLI runs GPT-5.6 with ultra reasoning to produce the diagnosis, evidence, bounded plan, selected file, exact edits, requested tests, diff explanation, and risk. GPT-5.6 cannot apply the patch, choose commands, approve a release, or deploy. BOB Forge records the concrete provider, model, reasoning effort, transport, billing mode, and whether any degraded path was used. Challenges and lessons The hardest challenge was not reconnect logic. It was maintaining one trustworthy identity for the candidate across analysis, validation, approval, deployment, rollback, and export. That led to immutable snapshots, digest-bound approvals, fail-closed state transitions, and a portable verifier that recomputes the evidence verdict. I also learned that a truthful local-first product needs to survive reloads and long model turns without faking progress. Durable events, resumable state, explicit readiness, and visible failure states became core product features rather than polish. Build Week disclosure BOB Forge is a separate repository created during Build Week. Its first provenance commit, 52131cd, preserves a visual HUD foundation imported from my pre-existing Hermes OS project. That visual language, animation system, responsive layout foundation, and service-worker foundation are not claimed as new work. The BOB-specific control plane, local Codex boundary, isolated Git workspaces, validation engine, Docker gates, phone approval workflow, deployment and rollback system, evidence format, portable verifier, and product workflow were built during Build Week. The repository contains the detailed boundary and dated commit history. Current scope The submission intentionally proves one deep vertical slice: reconnect-demo, one permitted source file, and a local Docker target. It does not claim arbitrary-repository autonomy or production deployment. The rollback rehearsal restores the exact registered broken baseline to prove reversibility; it is not presented as a previous healthy production release. BOB Forge is built around one principle: an AI-generated patch should not automatically become a release. ## README (from the GitHub repository) # BOB Forge **BOB Forge turns an issue into a verified fix.** ![BOB Forge local-first repair and release flow](docs/assets/bob-forge-flow.svg) BOB Forge is for solo developers and small teams that do not have dedicated SRE or DevOps staff. It is a local-first Build Week engineering demo that inspects one registered read-only project, asks a structured execution provider for a bounded repair proposal, copies the source into an isolated local Git workspace, validates and tests the change, then waits at one post-test human approval gate. An approved artifact can target only the local Docker sandbox. There is no VPS, hosted control plane, production deployment, source merge, or push path. **Why not just Codex?** Codex is the repair intelligence and proposal engine; BOB Forge is the deterministic control plane around isolation, patch validation, testing, human approval, portable evidence, local deployment, health verification, and rollback. The model proposes a repair, while BOB Forge owns what may run and whether the tested release may deploy. > **Honest hackathon scope:** the current product supports one bundled target, `reconnect-demo`. It repairs a real intentionally broken WebSocket reconnect fixture; it is not yet a generic arbitrary-repository agent or a production deployer. ## Judge demo path ### 1. Watch the product The public demo URL is supplied in the Devpost entry. It shows the real `reconnect-demo` repair flow, including local Codex/GPT-5.6 analysis, isolated Docker gates, phone approval, measured health, explicit rollback, and the final evidence verdict. ### 2. Verify the preserved run The final redacted Evidence Pack can be checked without Docker, a phone, an API key, or access to the developer's Codex login: ```powershell .\scripts\verify-evidence.ps1 C:\path\to\bob-forge-evidence-task_id.json ``` The verifier recomputes the canonical-payload self-consistency digest, stable run digest, and semantic verdict, then exits non-zero on a mismatch. Judge Mode also embeds the final application commit. This portable path is evidence verification, not a simulated model run or a hosted demo; see [Judge test path](docs/JUDGE_TEST_PATH.md) for the trust boundary. ### 3. Run a complete local repair From the repository root, run the clean-checkout preflight once. It validates the local ChatGPT Codex login, the project-pinned local Docker engine, tests, linters, reset, and fixture warmup without starting the server: ```powershell .\scripts\preflight.ps1 ``` After committing the exact bytes under review, create a real local-Codex/Docker proof with: ```powershell .\scripts\run-e2e.ps1 ``` For the timed run, keep the PC and phone on the same trusted Wi-Fi and start: ```powershell .\scripts\judge-demo.ps1 -LaunchOnly ``` Open the newly printed `/pair` URL on the phone, enter that restart's code, submit the prefilled `reconnect-demo` incident, inspect the isolated diff and Docker test evidence, approve the exact artifact, show reconnect health, roll back to the measured broken baseline, and tap **Export evidence pack**. The current export is schema v2: require `verdict.status = "VERIFIED"` and `verdict.verified = true`; `INCOMPLETE` or `MISMATCH` is evidence of a missing or contradictory proof, not a passing demo. The concise operator flow is in [Judge quickstart](docs/JUDGE_QUICKSTART.md), and the full cue sheet is in [docs/JUDGING.md](docs/JUDGING.md). If a physical phone is unavailable, use the same HUD on loopback and say so; never weaken the access checks. ## Quick start Requirements: Python 3.12+, PowerShell, Git, an installed Codex CLI signed in through ChatGPT, and a running local Docker Desktop engine for executable validation/deployment. Node.js is optional and is used only for frontend syntax checks. Setup now verifies the project-resolved local Docker endpoint and a real engine response; Docker absence fails before model-authored code can run, with no host or remote-daemon fallback. The default local Codex path uses the user's eligible ChatGPT Codex plan, explicitly pins Build Week runs to `gpt-5.6-sol` with `ultra` reasoning, and does not require an OpenAI API key. ```powershell Set-Location -LiteralPath '' .\scripts\setup.ps1 .\scripts\start-demo.ps1 ``` Open `http://127.0.0.1:8138`, select `reconnect-demo`, and submit the prefilled reconnect incident. Runtime state is stored under `.bob-forge/` and can be removed safely with: ```powershell .\scripts\reset-demo.ps1 ``` `start-demo.ps1` always runs the BOB Forge control plane as a Windows host process so it can use the local Codex and Docker CLIs. `docker-compose.yml` now launches only the intentionally broken `upstream` and `broken-demo` fixtures; it is not a BOB Forge server or deployment adapter. The root control-plane image was removed. Deployment builds the reviewed `demo_service/Dockerfile` separately from the content-addressed pre-change baseline and the frozen candidate that already passed Docker tests. Execution selection is explicit: 1. `local_codex` is the default and uses hardened direct `codex exec --json --sandbox read-only` with user config/rules ignored for the task boundary, explicitly passing `--model gpt-5.6-sol` and `model_reasoning_effort="ultra"`; the complete bounded event stream is rejected if it reports command, file-change, web-search, or tool actions; 2. the pinned beta SDK remains available only through explicit `BOB_FORGE_CODEX_TRANSPORT=sdk_opt_in`, because it does not provide an equivalent ignore-user-config/rules boundary; 3. deterministic fallback is disabled by default; only a prepared non-judge `reconnect-demo` may opt in with the exact `BOB_FORGE_ALLOW_DETERMINISTIC_FALLBACK=YES` setting; 4. `openai_api` is a separate paid opt-in requiring `BOB_FORGE_EXECUTION_PROVIDER=openai_api`, `OPENAI_API_KEY`, and `BOB_FORGE_CONFIRM_PAID_API=YES`. Judge Mode forces `BOB_FORGE_ALLOW_DETERMINISTIC_FALLBACK=NO` and fails closed if local Codex is unavailable. There is never an automatic provider switch or paid-API fallback; an API key alone does not activate paid mode. See [execution and cost](docs/EXECUTION_AND_COST.md) and [OpenAI integration](docs/OPENAI.md). `BOB_FORGE_CODEX_MODEL` and `BOB_FORGE_CODEX_REASONING_EFFORT` can override those local defaults with validated values. `/api/system/status` reports the actual configured model and effort so the HUD can show what will run rather than inheriting an invisible global Codex setting. ### Optional paired phone view Loopback is always the default. On a trusted same-WiFi network, an explicit exact RFC1918 address enables paired phone access: ```powershell .\scripts\start-demo.ps1 -LanAddress 192.168.1.25 # lan-readonly .\scripts\start-demo.ps1 -LanAddress 192.168.1.25 -AllowLanControl # lan-control ``` The script verifies that the address is assigned to this PC on a Windows `Private` network profile, prints a restart-scoped `/pair` URL/code, and creates no firewall rule. LAN mode uses plain HTTP: use it only on trusted WiFi. The pairing cookie protects static/API/SSE access; control mode additionally uses exact Origin checks and session-bound CSRF. No HTTPS tunnel or VPS path is implemented or enabled by default. ## What is real - SQLite tasks, versioned transitions, events, commands, artifacts, approvals, deployments, and crash-boundary recovery; - durable SSE replay through `Last-Event-ID`, reload recovery, periodic persisted `execution_provider / working` heartbeats while local Codex is analyzing, and exact same-origin local HTTP enforcement; - bounded repository inspection, a fresh isolated Git baseline per task, provider-mutation detection including `.git`, a client-only edit allowlist, AST checks, and real pre-approval tests inside hardened local Docker; - one post-test release approval bound to the current task version, diagnosis/plan evidence, immutable pre-change baseline, tested candidate, and canonical deployment manifest; bo [README truncated for size] ## Detected evidence (automated analysis) Indexed codebase: 76 recognized source files, 1224 KB. - CSS (language) — detected in the code - HTML (language) — detected in the code - JavaScript (language) — detected in the code - Python (language) — detected in the code - Docker (technology) — claimed on Devpost, not found in the code ## Codebase structure (from repository index) ### Files (101 of 101) ``` .dockerignore .env.example .gitattributes .github/workflows/ci.yml .gitignore bob_forge/__init__.py bob_forge/__main__.py bob_forge/ai.py bob_forge/deployment.py bob_forge/evidence.py bob_forge/models.py bob_forge/orchestrator.py bob_forge/repository.py bob_forge/risk.py bob_forge/runner.py bob_forge/security.py bob_forge/server.py bob_forge/state_machine.py bob_forge/store.py bob_forge/workspace.py demo_service/Dockerfile demo_service/fixtures/incident.txt demo_service/README.md demo_service/reconnect_demo/__init__.py demo_service/reconnect_demo/app.py demo_service/reconnect_demo/client.py demo_service/reconnect_demo/upstream.py demo_service/requirements.txt demo_service/tests/test_reconnect.py docker-compose.yml docs/ARCHITECTURE.md docs/BUILD_WEEK_SCOPE.md docs/DECISIONS.md docs/DEMO_GUIDE.md docs/DEVPOST_SUBMISSION.md docs/EXECUTION_AND_COST.md docs/FINAL_COMPETITIVE_AUDIT.md docs/FINAL_GAP_AUDIT.md docs/FINAL_TEST_REPORT.md docs/HUD_AUDIT.md docs/JUDGE_QUICKSTART.md docs/JUDGE_TEST_PATH.md docs/JUDGING.md docs/KNOWN_LIMITATIONS.md docs/MANUAL_PHONE_QA.md docs/media/voiceover-en.srt docs/media/voiceover-en.txt docs/OPENAI.md docs/PRIOR_WORK.md docs/PROGRESS.md docs/ROADMAP_RECONCILIATION.md docs/SECURITY_MODEL.md docs/SETUP.md docs/SUBMISSION_CHECKLIST.md docs/TESTING.md docs/TROUBLESHOOTING.md docs/UI_REUSE.md docs/VIDEO_SCRIPT_3_MIN.md docs/VIDEO_VOICEOVER_2M50.md docs/VOICEOVER_PRODUCTION.md frontend/hud/forge.css frontend/hud/forge.js frontend/hud/index.html frontend/hud/manifest.webmanifest frontend/hud/sw.js LICENSE pyproject.toml README.md requirements-dev.txt requirements.txt scripts/check.ps1 scripts/judge-demo.ps1 scripts/preflight.ps1 scripts/prove-demo.ps1 scripts/prove-demo.py scripts/reset-demo.ps1 scripts/run-e2e.ps1 scripts/setup.ps1 scripts/start-demo.ps1 scripts/test.ps1 scripts/verify-broken-demo.ps1 scripts/verify-evidence.ps1 scripts/verify-evidence.py tests/__init__.py tests/test_ai.py tests/test_deployment.py tests/test_evidence_cli.py tests/test_evidence.py tests/test_frontend.py tests/test_orchestrator.py tests/test_release_assets.py tests/test_repository.py tests/test_risk.py tests/test_runner.py tests/test_scripts.py tests/test_security.py tests/test_server.py tests/test_state_machine.py tests/test_store.py tests/test_workspace.py THIRD_PARTY_NOTICES.md ``` ### Dependencies - demo_service/requirements.txt: websockets@==15.0.1 - requirements.txt: openai-codex@==0.1.0b3 ### Recent commits (newest first) - docs: replace ambiguous flow arrows with numbered layout - fix: improve desktop judge readability - ci: install bundled demo dependencies - ci: install pinned verification tools - fix: make release verification portable - polish: finalize public submission readiness - docs: freeze the submission candidate story - docs: package the Build Week judge story - feat: sharpen the mobile judge outcome - feat: add portable commit-bound evidence verification - security: enforce read-only Codex proposal turns - docs: make bounded analysis wait explicit - feat: show bounded Codex analysis time - fix: align interactive Codex timeout with proof window - docs: sync final phone failure verification facts - fix: keep failure guidance evidence-bound - fix: make provider capacity failures retryable - fix: ignore expected client disconnect errors - docs: record final release gates - docs: make release commands canonical ## Key source files (fetched from GitHub, selected and truncated for size) ### THIRD_PARTY_NOTICES.md ```markdown # Third-party notices BOB Forge is distributed under the [MIT License](LICENSE). Its source repository does not vendor Python dependency source or Codex authentication material. Runtime and development dependencies retain their own licenses: - `openai-codex==0.1.0b3` — Apache License 2.0. - `websockets==15.0.1` in the bundled reconnect fixture — BSD 3-Clause License. - Python, Git, Node.js, Docker Desktop, the Codex CLI, and browser runtimes are external prerequisites and are not redistributed by this repository. The pre-existing Hermes OS visual foundation reused by BOB Forge is disclosed in [Prior work](docs/PRIOR_WORK.md), [UI reuse](docs/UI_REUSE.md), and the baseline commit identified there. The BOB Forge repository license covers the submitted source while preserving the provenance disclosure. ``` ### docs/VIDEO_SCRIPT_3_MIN.md ```markdown # Legacy three-minute video outline This older outline is retained only for Build Week audit history. Do not record from it: the submission rule requires a video **strictly under three minutes**, and its former `3:00` endpoint left no safe margin. Use the canonical [2:48 voiceover and shot list](VIDEO_VOICEOVER_2M50.md), the [AI voiceover production handoff](VOICEOVER_PRODUCTION.md), and [Judge quickstart](JUDGE_QUICKSTART.md). They preserve the same real-run, single-approval, measured-health, explicit-rollback, redaction, and honest-scope requirements with a twelve-second safety margin. ``` ### requirements.txt ``` openai-codex==0.1.0b3 ``` ### pyproject.toml ``` [tool.ruff] target-version = "py312" line-length = 120 src = ["bob_forge", "demo_service/reconnect_demo", "tests", "demo_service/tests"] exclude = [".bob-forge", ".venv"] [tool.ruff.lint] select = ["E", "F", "I", "UP", "B", "SIM", "RUF"] ignore = [ "E501", # Existing long literals are clearer left intact; the formatter handles code layout. "RUF001", # The repository scanner intentionally includes Cyrillic letters. "RUF100", # Keep protocol-required method-name nosec/noqa markers locally explicit. "UP042", # Preserve the established str-enum runtime representation. "UP047", # TypeVar syntax keeps the package usable by pre-3.12 type tooling. ] [tool.ruff.lint.isort] known-first-party = ["bob_forge", "reconnect_demo"] [tool.mypy] python_version = "3.12" files = ["bob_forge", "demo_service/reconnect_demo"] strict = true warn_unreachable = true pretty = true show_error_codes = true exclude = ["^\\.bob-forge/", "^\\.venv/"] ``` ### docker-compose.yml ```yaml name: bob-forge-build-week services: upstream: build: context: demo_service dockerfile: Dockerfile command: ["python", "-m", "reconnect_demo.upstream", "--host", "0.0.0.0", "--port", "8765"] expose: - "8765" healthcheck: test: ["CMD", "python", "-c", "from pathlib import Path; rows = Path('/proc/net/tcp').read_text().splitlines()[1:]; assert any(row.split()[1].endswith(':223D') and row.split()[3] == '0A' for row in rows)"] interval: 1s timeout: 3s retries: 15 start_period: 1s read_only: true tmpfs: - /tmp:size=32m,noexec,nosuid security_opt: - no-new-privileges:true cap_drop: - ALL pids_limit: 64 mem_limit: 128m cpus: 0.5 broken-demo: build: context: demo_service dockerfile: Dockerfile environment: UPSTREAM_URL: ws://upstream:8765 ports: - "127.0.0.1:8081:8081" depends_on: upstream: condition: service_healthy read_only: true tmpfs: - /tmp:size=32m,noexec,nosuid security_opt: - no-new-privileges:true cap_drop: - ALL pids_limit: 64 mem_limit: 128m cpus: 0.5 ``` ### demo_service/requirements.txt ``` websockets==15.0.1 ``` ### demo_service/Dockerfile ``` FROM python:3.12-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt RUN groupadd --system forge && useradd --system --gid forge --home /app forge COPY --chown=forge:forge reconnect_demo ./reconnect_demo ENV PYTHONUNBUFFERED=1 USER forge CMD ["python", "-m", "reconnect_demo.app"] ``` ### demo_service/reconnect_demo/app.py ```python from __future__ import annotations import asyncio import json import os import threading from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from typing import Any, TypedDict from .client import FeedConsumer class ServiceState(TypedDict, total=False): status: str connected: bool messages_received: int last_sequence: int | None reconnect_count: int error: str STATE: ServiceState = { "status": "starting", "connected": False, "messages_received": 0, "last_sequence": None, "reconnect_count": 0, } def consume() -> None: def state(value: str) -> None: STATE["status"] = value STATE["connected"] = value == "connected" def message(payload: dict[str, Any]) -> None: STATE["messages_received"] += 1 STATE["last_sequence"] = payload.get("sequence") def reconnect() -> None: STATE["reconnect_count"] += 1 async def work() -> None: consumer = FeedConsumer( os.getenv("UPSTREAM_URL", "ws://upstream:8765"), on_state=state, on_message=message, on_reconnect=reconnect, ) await consumer.collect(1_000_000) try: asyncio.run(work()) except Exception as exc: # the health process deliberately survives the worker STATE["status"] = "stalled" STATE["error"] = f"{type(exc).__name__}: {exc}" class HealthHandler(BaseHTTPRequestHandler): def do_GET(self) -> None: # noqa: N802 if self.path not in {"/", "/health"}: self.send_error(404) return body = json.dumps(STATE, separators=(",", ":")).encode() self.send_response(200) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def main() -> None: threading.Thread(target=consume, daemon=True).start() ThreadingHTTPServer(("0.0.0.0", 8081), HealthHandler).serve_forever() if __name__ == "__main__": main() ``` ### bob_forge/server.py ```python from __future__ import annotations import argparse import hashlib import hmac import ipaddress import json import os import re import secrets import sys import threading import time from dataclasses import dataclass, field from datetime import UTC, datetime from enum import StrEnum from http import HTTPStatus from http.cookies import CookieError, SimpleCookie from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from typing import Any, cast from urllib.parse import unquote, urlparse from .ai import AIAdapter from .deployment import DockerSandbox, DockerUnavailable from .evidence import build_judge_evidence from .models import ConflictError, RecordNotFound, VersionConflict from .orchestrator import DEFAULT_INCIDENT_LOG, ForgeOrchestrator from .repository import PROJECT_REGISTRY from .security import redact_text, redact_value, safe_static_path from .store import SQLiteStore from .workspace import WorkspaceManager ROOT = Path(__file__).resolve().parents[1] DEFAULT_HUD = ROOT / "frontend" / "hud" DEFAULT_DATA = ROOT / ".bob-forge" MAX_BODY_BYTES = 64 * 1024 _TASK_ROUTE = re.compile(r"^/api/tasks/([^/]+)$") _EVIDENCE_ROUTE = re.compile(r"^/api/tasks/([^/]+)/evidence$") _APPROVAL_ROUTE = re.compile(r"^/api/tasks/([^/]+)/approval$") _ROLLBACK_ROUTE = re.compile(r"^/api/tasks/([^/]+)/rollback$") _LOOPBACK_HOSTS = frozenset({"127.0.0.1", "localhost", "::1"}) _RFC1918_NETWORKS = tuple( ipaddress.ip_network(cidr) for cidr in ("10.0.0.0/8", "172.16.0.0/12", "192.168.0.0/16") ) _SESSION_COOKIE = "bob_forge_session" _CSRF_HEADER = "X-BOB-Forge-CSRF" _SESSION_LIFETIME_S = 8 * 60 * 60 _MAX_LIVE_SESSIONS = 64 _SHUTDOWN_IDLE_WAIT_S = 30.0 _PAIR_FAILURE_LIMIT = 5 _PAIR_FAILURE_WINDOW_S = 60.0 _GIT_COMMIT_RE = re.compile(r"^[0-9a-f]{40}(?:[0-9a-f]{24})?$") class AccessMode(StrEnum): LOOPBACK = "loopback" LAN_READONLY = "lan-readonly" LAN_CONTROL = "lan-control" class RuntimeUnavailable(RuntimeError): """The durable runtime is not yet safe to mutate.""" def _is_rfc1918_ipv4(value: str) -> bool: try: address = ipaddress.ip_address(value) except ValueError: return False return isinstance(address, ipaddress.IPv4Address) and any(address in network for network in _RFC1918_NETWORKS) def _canonical_pairing_code(value: str) -> str: """Accept harmless grouping/case differences without putting the secret in a URL.""" return re.sub(r"[\s-]+", "", value).upper() def _validate_access_request(host: str, mode: AccessMode, pairing_secret: str | None) -> None: if mode is AccessMode.LOOPBACK: if host.lower() not in _LOOPBACK_HOSTS: raise ValueError("non-loopback bind requires explicit lan-readonly or lan-control access mode") return if not _is_rfc1918_ipv4(host): raise ValueError("LAN mode requires one exact RFC1918 IPv4 address; wildcard/public binds are refused") if ( pairing_secret is None or len(pairing_secret) < 32 or len(_canonical_pairing_code(pairing_secret)) < 20 ): raise ValueError( "LAN mode requires BOB_FORGE_PAIRING_SECRET with at least 32 grouped characters " "and 20 non-separator symbols" ) @dataclass(frozen=True) class AccessPolicy: mode: AccessMode bind_host: str port: int pairing_digest: bytes | None = field(default=None, repr=False) @classmethod def create( cls, *, mode: AccessMode, bind_host: str, port: int, pairing_secret: str | None, ) -> AccessPolicy: _validate_access_request(bind_host, mode, pairing_secret) digest = ( None if mode is AccessMode.LOOPBACK else hashlib.sha256(_canonical_pairing_code(pairing_secret or "").encode()).digest() ) return cls(mode=mode, bind_host=bind_host.lower(), port=port, pairing_digest=digest) @property def auth_required(self) -> bool: return self.mode is not AccessMode.LOOPBACK @property def control_enabled(self) -> bool: return self.mode in {AccessMode.LOOPBACK, AccessMode.LAN_CONTROL} @property def authority(self) -> str: return f"{self.bind_host}:{self.port}" def allows_host(self, raw_host: str) -> bool: if self.mode is AccessMode.LOOPBACK: return _host_name(raw_host) in _LOOPBACK_HOSTS return raw_host.lower() == self.authority def allows_origin(self, origin: str) -> bool: if self.mode is AccessMode.LOOPBACK: try: parsed = urlparse(origin) except ValueError: return False return ( parsed.scheme in {"http", "https"} and (parsed.hostname or "").lower() in _LOOPBACK_HOSTS and parsed.netloc.lower() != "" ) return origin.lower() == f"http://{self.authority}" def verifies_pairing_code(self, code: str) -> bool: if self.pairing_digest is None: return False supplied = hashlib.sha256(_canonical_pairing_code(code).encode()).digest() return hmac.compare_digest(self.pairing_digest, supplied) def public_status(self) -> dict[str, Any]: return { "mode": self.mode.value, "auth_required": self.auth_required, "control_enabled": self.control_enabled, "authority": self.authority, } @dataclass(frozen=True) class AccessSession: session_id: str csrf_token: str actor: str created_monotonic: float expires_monotonic: float def active_at(self, now: float) -> bool: return self.created_monotonic <= now < self.expires_monotonic def utc_now() -> str: return datetime.now(UTC).isoformat() def _application_commit_from_environment() -> str | None: value = os.getenv("BOB_FORGE_APPLICATION_COMMIT", "").strip().lower() return value if _GIT_COMMIT_RE.fullmatch [truncated — 40928 more characters] ``` ### tests/__init__.py ```python ``` [62 more indexed source files omitted to keep this export small. The full file list is in the Codebase structure section above.]