mortenVChristiansen
/
k_card
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Add CTAP probe and update phase docs

This commit is contained in:
Morten V. Christiansen 2026-04-25 10:25:40 +02:00
parent d0d27a0896
commit 2448956946
8 changed files with 1507 additions and 68 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

48
PHASE5_RUNBOOK.md
View File

@ -150,9 +150,57 @@ curl -i --cacert /home/user/chromecard/tls/phase2/ca.crt -X POST https://127.0.0
-H "Authorization: Bearer $TOKEN" -H "Authorization: Bearer $TOKEN"
``` ```
## Regression Script
For the split-VM chain, use the host-side regression helper:
```bash
/home/user/chromecard/phase5_chain_regression.sh
```
Defaults:
- Drives the test from `k_client` over SSH.
- Uses `https://127.0.0.1:9771` and `/home/user/chromecard/tls/phase2/ca.crt` inside `k_client`.
- Logs in as `alice`.
- Runs `20` counter requests at parallelism `8`.
- Verifies that returned counter values are unique and gap-free, then logs out and checks for `401` after logout.
Useful overrides:
```bash
REQUESTS=50 PARALLELISM=12 /home/user/chromecard/phase5_chain_regression.sh
```
```bash
/home/user/chromecard/phase5_chain_regression.sh --username alice --client-host k_client
```
Verified result on 2026-04-25:
- Live split-VM chain passed end-to-end.
- Login, session status, counter reuse, and logout all worked from `k_client`.
- A `20` request / `8` worker concurrency burst returned unique, gap-free counter values `23..42`.
## Current Limitation ## Current Limitation
- This uses card-presence probing, not a full WebAuthn assertion verification path. - This uses card-presence probing, not a full WebAuthn assertion verification path.
- Intended as a Phase 5 starter for session semantics and proxy/server behavior. - Intended as a Phase 5 starter for session semantics and proxy/server behavior.
- Session and counter state are currently process-local only; restart loses state. - Session and counter state are currently process-local only; restart loses state.
- Upstream trust still relies on a shared static `X-Proxy-Token`. - Upstream trust still relies on a shared static `X-Proxy-Token`.
- Experimental direct FIDO2 mode now exists in `k_proxy_app.py` behind `--auth-mode fido2-direct`, but it is not the default runtime:
- direct registration on the current `k_proxy` card/library stack still fails with `No compatible PIN/UV protocols supported!`
- a CTAP1 fallback probe did not complete quickly enough to promote as the working path
- the deployed service was restored to default `probe` mode so the validated Phase 5 chain remains usable
- Raw CTAP debugging helper now exists at `/home/user/chromecard/raw_ctap_probe.py`:
- use it on `k_proxy` to exercise low-level `makeCredential` / `getAssertion`
- it logs keepalive callbacks and transport exceptions
- Current blocker before the next direct-auth attempt:
- `k_proxy` currently has no visible `/dev/hidraw*`
- `python3 /home/user/chromecard/fido2_probe.py --list` in `k_proxy` returns `No CTAP HID devices found.`
- restore card visibility first, then retry the raw CTAP probe and stop to tell the user when to press `yes` or `no`
- Latest retry after card reattach:
- `/dev/hidraw0` and `/dev/hidraw1` are visible in `k_proxy` again
- `/dev/hidraw0` opens successfully as the normal user, but `/dev/hidraw1` is still permission-denied
- raw `makeCredential` still shows no card prompt, so the hang is before the firmware confirmation UI
- next step is to identify which hidraw interface `python-fido2` is selecting

134
Setup.md
View File

@ -359,6 +359,72 @@ Session note (2026-04-25, browser target moved to k_proxy):
- browser traffic is now intended to go straight to `k_proxy` - browser traffic is now intended to go straight to `k_proxy`
- the `k_client` portal remains only as a temporary bridge/compatibility layer - the `k_client` portal remains only as a temporary bridge/compatibility layer
Session note (2026-04-25, provisional enrollment hardening):
- The enrollment contract in `k_proxy` is now explicit but provisional.
- Current prototype enrollment rules:
- usernames are canonicalized to lowercase
- allowed username pattern is `3-32` chars using lowercase letters, digits, `.`, `_`, `-`
- optional `display_name` is allowed up to `64` chars
- enrollment create is create-only and duplicate create returns `user already enrolled`
- enrollment update is a separate operation
- enrollment delete is a separate operation and removes any active sessions for that username
- Current enrollment endpoints on `k_proxy`:
- `POST /enroll/register`
- `GET /enroll/status?username=<name>`
- `POST /enroll/update`
- `POST /enroll/delete`
- `GET /enroll/list`
- Verified behavior from `k_client` against `https://127.0.0.1:9771`:
- invalid username `A!` is rejected
- create for `dave` with `display_name` succeeds
- duplicate create for `dave` is rejected
- update for `dave` succeeds
- list returns enrolled users and metadata
- delete for `dave` succeeds
- login for deleted `dave` fails with `user not enrolled`
- Deliberate current limit:
- enrollment itself still does not require card presence; only login does
- this was kept lightweight because the enrollment semantics are expected to change later
Session note (2026-04-25, Phase 6.5 concurrency probe):
- Added reproducible concurrency probe:
- `/home/user/chromecard/phase65_concurrency_probe.py`
- probe now supports `--max-workers` so client-side fan-out can be swept explicitly
- Successful baseline run from `k_client` against direct proxy path:
- `3` users
- `4` protected requests per user
- `12/12` requests succeeded
- counter values were unique and contiguous from `6` to `17`
- max observed latency was about `457 ms`
- Larger follow-up run exposed current limit:
- `5` users
- `5` protected requests per user
- `18/25` requests succeeded
- failures returned TLS EOF / upstream unavailable errors
- successful counter values were still unique and contiguous from `18` to `35`
- max observed latency was about `758 ms`
- Additional Phase 6.5 diagnosis:
- fixed a keep-alive/body-drain bug in the HTTP/1.1 experiment so `k_server` no longer misparses follow-on requests as `{}POST`
- added an upstream connection pool in `k_proxy`; current default/test setting clamps `k_proxy -> k_server` to one pooled TLS connection
- despite that change, a full fan-out run with `25` in-flight protected calls still fails on client-observed TLS EOFs
- a worker-limited run now passes cleanly:
- `5` users
- `5` protected requests per user
- `25/25` requests succeeded with `--max-workers 10`
- raising client-side fan-out still breaks:
- `22/25` requests succeeded with `--max-workers 15`
- `15/25` requests succeeded with fully unbounded `25` workers in the latest rerun
- Current diagnosis:
- the protected counter and session logic stay correct under load; successful values remain unique and contiguous
- `k_proxy` and `k_server` can complete the requests that actually reach them
- the primary collapse point in current testing is the client-facing Qubes forwarder on `9771`
- `qvm_connect_9771.log` shows `qrexec-agent-data` / data-vchan failures and repeated `xs_transaction_start: No space left on device`
- `qvm_connect_9780.log` also showed earlier qrexec failures, but the latest worker-threshold evidence points first to connection fan-out on `k_client -> k_proxy`
- Practical meaning:
- the application logic is good for moderate concurrent use in the current prototype
- the direct browser path appears stable around `10` in-flight protected calls in the current Qubes setup
- the current concurrency ceiling is being set by Qubes forwarding behavior rather than by the monotonic counter logic
Session note (2026-04-25, in-VM forwarding test): Session note (2026-04-25, in-VM forwarding test):
- Tested the intended in-VM forwarding path with `qvm-connect-tcp` instead of host-side `qrexec-client-vm`. - Tested the intended in-VM forwarding path with `qvm-connect-tcp` instead of host-side `qrexec-client-vm`.
- Forwarders start and bind locally: - Forwarders start and bind locally:
@ -428,6 +494,61 @@ Session note (2026-04-25, dom0 policy fix validated):
- `k_client -> k_proxy -> k_server` chain is operational - `k_client -> k_proxy -> k_server` chain is operational
- session reuse and logout behavior are working in the current prototype - session reuse and logout behavior are working in the current prototype
Session note (2026-04-25, live chain re-validation and regression helper):
- Re-validated the split-VM chain after restart using the current TLS/localhost-forward shape:
- `k_client` local `9771` -> `k_proxy:8771`
- `k_proxy` local `9780` -> `k_server:8780`
- Verified live service state during this run:
- `k_server` local `https://127.0.0.1:8780/health` returned `ok=true`
- `k_proxy` local `https://127.0.0.1:8771/health` returned `ok=true`
- `k_proxy` local `https://127.0.0.1:9780/health` reached `k_server`
- `k_client` local `https://127.0.0.1:9771/health` reached `k_proxy`
- Verified end-to-end behavior from `k_client`:
- login for `alice` succeeded
- session status succeeded
- protected counter calls succeeded with session reuse
- logout succeeded
- post-logout protected access returned `401 invalid or expired session`
- Added reproducible regression helper at:
- `/home/user/chromecard/phase5_chain_regression.sh`
- Verified the new helper end-to-end on 2026-04-25:
- default run uses `20` requests at parallelism `8`
- returned values were unique and gap-free
- latest verified counter range from the helper was `43..62`
- Practical meaning:
- the current blocker is no longer Qubes forwarding for the base Phase 5 chain
- the current next-step gap is auth semantics, not transport bring-up
Session note (2026-04-25, direct FIDO2 auth attempt):
- Added an experimental direct FIDO2 path in `/home/user/chromecard/k_proxy_app.py`:
- runtime switch: `--auth-mode fido2-direct`
- default runtime remains `probe`
- Added a low-level CTAP helper at `/home/user/chromecard/raw_ctap_probe.py`:
- purpose: bypass `Fido2Client` and exercise raw CTAP2 `makeCredential` / `getAssertion`
- logs keepalive callbacks and exact transport exceptions for host-side debugging
- Direct-mode intent:
- replace the legacy `fido2_probe.py --json` session gate
- perform real credential registration and real assertion verification locally in `k_proxy` with `python-fido2`
- Current observed blocker on `k_proxy`:
- direct `make_credential` fails with `No compatible PIN/UV protocols supported!`
- reproduces outside the app in a minimal VM-side probe, so this is not just a handler bug
- likely cause is the current card / `python-fido2` stack selecting a PIN/UV-dependent CTAP2 path for registration
- Additional probe:
- a forced CTAP1 fallback experiment did not fail immediately, but also did not complete quickly enough to treat as a usable working path in this turn
- Latest live blocker (2026-04-25, after refactor/deploy):
- direct probing is currently blocked before the card Yes/No UI stage because `k_proxy` no longer sees any CTAP HID device
- `ssh k_proxy "python3 /home/user/chromecard/fido2_probe.py --list"` now returns `No CTAP HID devices found.`
- `ssh k_proxy "ls -l /dev/hidraw*"` shows no `hidraw` nodes at the moment
- Follow-up after card reattach (2026-04-25):
- `k_proxy` again shows `/dev/hidraw0` and `/dev/hidraw1`
- direct node-open check confirms `/dev/hidraw0` is readable as the normal user
- `/dev/hidraw1` still returns `PermissionError: [Errno 13] Permission denied`
- raw `makeCredential` probe still produced no on-card registration prompt, so the host path is hanging before the firmware Yes/No UI
- Practical outcome for this session:
- the experimental direct mode is kept in code for follow-up work
- the deployed `k_proxy` service was restored to default `probe` mode
- verified `alice` login still works afterward, so the validated Phase 5 baseline remains intact
## Known FIDO2 Transport Boundary ## Known FIDO2 Transport Boundary
- FIDO2 on this firmware is handled via USB HID (CTAPHID), not Wi-Fi/BLE/MQTT. - FIDO2 on this firmware is handled via USB HID (CTAPHID), not Wi-Fi/BLE/MQTT.
@ -453,6 +574,9 @@ SUBSYSTEM=="hidraw", ATTRS{idVendor}=="1209", ATTRS{idProduct}=="0005", MODE="06
- `python3 /home/user/chromecard/fido2_probe.py --list` - `python3 /home/user/chromecard/fido2_probe.py --list`
- Then: - Then:
- `python3 /home/user/chromecard/fido2_probe.py --json` - `python3 /home/user/chromecard/fido2_probe.py --json`
- For raw CTAP debugging on `k_proxy`:
- `python3 /home/user/chromecard/raw_ctap_probe.py info`
- `python3 /home/user/chromecard/raw_ctap_probe.py make-credential --rp-id localhost`
4. Run local WebAuthn bring-up demo. 4. Run local WebAuthn bring-up demo.
- `python3 /home/user/chromecard/webauthn_local_demo.py` - `python3 /home/user/chromecard/webauthn_local_demo.py`
@ -483,8 +607,16 @@ SUBSYSTEM=="hidraw", ATTRS{idVendor}=="1209", ATTRS{idProduct}=="0005", MODE="06
- Whether a full `CR_SDK_CK-main` checkout (with role directories) is available locally. - Whether a full `CR_SDK_CK-main` checkout (with role directories) is available locally.
- Whether server-side code should be pulled now for broader CIP/WebAuthn integration testing. - Whether server-side code should be pulled now for broader CIP/WebAuthn integration testing.
- Exact Qubes firewall and service binding rules to enforce the `k_client -> k_proxy -> k_server` chain.
- Exact enrollment process interface running in `k_client` and how it reaches `k_proxy`. - Exact enrollment process interface running in `k_client` and how it reaches `k_proxy`.
- Upgrade Phase 5 auth gate from card-presence probe to full WebAuthn assertion verification for session creation. - Upgrade Phase 5 auth gate from card-presence probe to full WebAuthn assertion verification for session creation.
- Determine the viable path for real credential registration on `k_proxy`:
- enable whatever PIN/UV support the card expects for direct CTAP2 registration, or
- adopt a different one-time enrollment path that can persist real credential material for later direct assertion verification.
- Restore card visibility inside `k_proxy` so direct probes can reach the card UI again:
- `/dev/hidraw*` must exist in `k_proxy`
- `fido2_probe.py --list` must detect the card before the raw Yes/No probe can continue
- Identify why the host probe hangs before card UI even with `/dev/hidraw0` readable:
- determine which hidraw interface `python-fido2` is selecting on `k_proxy`
- determine whether the blocked path is on the second HID interface or in the Qubes USB mediation layer
- Precise ownership split of session/user state between `k_proxy` and `k_server`. - Precise ownership split of session/user state between `k_proxy` and `k_server`.
- Concrete concurrency limits and acceptance criteria (requests/sec, parallel clients, latency/error thresholds). - Concrete concurrency limits and acceptance criteria (requests/sec, parallel clients, latency/error thresholds).

110
Workplan.md
View File

@ -42,7 +42,7 @@ This is the execution plan for making ChromeCard FIDO2 development and validatio
Exit criteria: Exit criteria:
- All 3 VMs exist, boot, and have clearly defined service ownership. - All 3 VMs exist, boot, and have clearly defined service ownership.
## Phase 1: Qubes Firewall Policy (Blocking) ## Phase 1: Qubes Firewall Policy
1. Enforce allowed forward paths only. 1. Enforce allowed forward paths only.
- Allow `k_client` outbound TLS only to `k_proxy` service port(s). - Allow `k_client` outbound TLS only to `k_proxy` service port(s).
@ -76,6 +76,16 @@ Status (2026-04-25, after restart and service recovery):
- Immediate next action for Phase 1: - Immediate next action for Phase 1:
- verify and fix the dom0 policy/mechanism that should permit `qubes.ConnectTCP` forwarding for the chain ports - verify and fix the dom0 policy/mechanism that should permit `qubes.ConnectTCP` forwarding for the chain ports
Status (2026-04-25, dom0 policy fix validated):
- The forwarding blocker is cleared for the current prototype shape.
- Verified working chain:
- `k_client` localhost `9771` -> `k_proxy:8771`
- `k_proxy` localhost `9780` -> `k_server:8780`
- Verified outcome:
- TLS health checks pass on both hops
- end-to-end login, session status, protected counter access, and logout all succeed from `k_client`
- Phase 1 is complete for the current localhost-forwarded `qubes.ConnectTCP` design.
## Phase 2: TLS Certificates and Service Endpoints ## Phase 2: TLS Certificates and Service Endpoints
1. Certificate model. 1. Certificate model.
@ -227,7 +237,23 @@ Status (2026-04-25):
- Current split-VM test shape is: - Current split-VM test shape is:
- `k_proxy` listening on `127.0.0.1:8771` - `k_proxy` listening on `127.0.0.1:8771`
- `k_server` listening on `127.0.0.1:8780` - `k_server` listening on `127.0.0.1:8780`
- Phase 5 application logic is runnable locally inside each VM, but end-to-end validation is still blocked by Phase 1 qrexec forwarding refusal. - End-to-end validation is now passing through the live chain from `k_client`.
- Current verified behavior:
- login succeeds for `alice`
- session status succeeds
- repeated protected counter requests succeed with session reuse
- logout succeeds
- post-logout protected access returns `401`
- Added repeatable host-side regression helper:
- `/home/user/chromecard/phase5_chain_regression.sh`
- Phase 5 is complete for the current prototype semantics.
- Experimental follow-up in code:
- `k_proxy_app.py` now also has `--auth-mode fido2-direct`
- this mode attempts direct credential registration and direct assertion verification with `python-fido2`
- it is not the deployed default because direct registration currently fails on `k_proxy` with `No compatible PIN/UV protocols supported!`
- `/home/user/chromecard/raw_ctap_probe.py` now exists for lower-level CTAP2 probing with keepalive/error logging
- latest retry result: after reattaching the card, `k_proxy` again exposes `/dev/hidraw0` and `/dev/hidraw1`, but raw `makeCredential` still reaches no Yes/No card prompt
- `/dev/hidraw0` opens successfully as the normal user; `/dev/hidraw1` is still permission-denied
## Phase 5.5: Implement Dummy Resource + Access Policy on `k_server` ## Phase 5.5: Implement Dummy Resource + Access Policy on `k_server`
@ -245,6 +271,14 @@ Exit criteria:
- Authorized requests obtain consistent increasing values. - Authorized requests obtain consistent increasing values.
- Unauthorized requests are rejected. - Unauthorized requests are rejected.
Status (2026-04-25):
- The protected counter resource is implemented and validated in the live split-VM chain.
- Verified behavior:
- authorized requests from `k_proxy` obtain increasing values
- unauthorized post-logout requests from `k_client` are rejected with `401`
- `20` concurrent protected requests through the chain returned unique, gap-free values
- Phase 5.5 is complete for the current prototype shape.
## Phase 6: Integrate Client Enrollment + Proxy Login Flow ## Phase 6: Integrate Client Enrollment + Proxy Login Flow
1. Enrollment process in `k_client`. 1. Enrollment process in `k_client`.
@ -257,6 +291,15 @@ Exit criteria:
3. Browser flow in `k_client`. 3. Browser flow in `k_client`.
- Browser traffic goes only to `k_proxy`. - Browser traffic goes only to `k_proxy`.
Immediate next action:
- Determine which hidraw interface the host CTAP stack is actually selecting on `k_proxy`.
- Verify which interface is blocked:
- map `/dev/hidraw0` and `/dev/hidraw1` to their USB/HID descriptors
- determine whether `python-fido2` is trying to use the permission-blocked interface
- Then retry:
- `ssh k_proxy "python3 /home/user/chromecard/raw_ctap_probe.py make-credential --rp-id localhost"`
- Stop before the raw probe and tell the user explicitly to press `yes` or `no` on the card.
- Validate end-to-end login to `k_server` resource through proxy chain. - Validate end-to-end login to `k_server` resource through proxy chain.
Exit criteria: Exit criteria:
@ -285,6 +328,61 @@ Status (2026-04-25):
- the `k_client` bridge remains only for transition/compatibility - the `k_client` bridge remains only for transition/compatibility
- final enrollment semantics are still provisional - final enrollment semantics are still provisional
Status (2026-04-25, enrollment hardening):
- Added a more explicit provisional enrollment contract in `k_proxy`:
- username normalization and validation
- optional `display_name`
- separate create, update, delete, status, and list operations
- delete invalidates existing sessions for that username
- Verified the hardened behaviors on the direct proxy path.
- Phase 6 is now strong enough to treat the browser/proxy flow as a stable prototype baseline.
- The remaining reason Phase 6 is not "final" is product semantics, not missing basic mechanics:
- whether enrollment should require card presence
- what user attributes belong in enrollment
- what re-enroll and recovery should mean
Status (2026-04-25, Phase 6.5 initial concurrency results):
- Added reproducible probe script at `/home/user/chromecard/phase65_concurrency_probe.py`.
- Probe now supports `--max-workers` so client-side fan-out can be tested separately from total request count.
- Moderate direct-path concurrency passes:
- `3 users x 4 requests`
- `12/12` successful protected calls
- counter values remained unique and contiguous
- Larger direct-path concurrency currently fails:
- `5 users x 5 requests`
- only `18/25` successful protected calls
- failed calls report TLS EOF / upstream unavailable errors
- Follow-up findings are more precise:
- body-drain handling was fixed for the HTTP/1.1 keep-alive experiment
- `k_proxy -> k_server` upstream concurrency is now clampable and currently tested at one pooled connection
- `5 users x 5 requests` passes at `25/25` when client fan-out is limited to `--max-workers 10`
- the same total load still fails at higher fan-out:
- `22/25` at `--max-workers 15`
- `15/25` at fully unbounded `25` workers in the latest rerun
- Current bottleneck is still not counter correctness:
- successful results still show unique, contiguous counter values
- `k_proxy` and `k_server` complete the requests that actually arrive
- Current likely bottleneck is the client-facing Qubes forwarding layer:
- `qvm_connect_9771.log` shows qrexec data-vchan failures
- observed message includes `xs_transaction_start: No space left on device`
- `qvm_connect_9780.log` showed earlier failures too, but the latest threshold test points first to connection fan-out on `k_client -> k_proxy`
- Phase 6.5 is therefore started but not complete:
- application-level concurrency looks acceptable at moderate load
- current working envelope is roughly `10` in-flight protected calls on the direct browser path
- higher-load failures still need Qubes forwarding diagnosis before the phase can be closed
Status (2026-04-25, Phase 5 regression helper):
- Added repeatable split-VM regression helper:
- `/home/user/chromecard/phase5_chain_regression.sh`
- Verified helper result on the live chain:
- `20` requests at parallelism `8`
- login/session-status/counter/logout sequence completed successfully
- returned counter values were unique and gap-free
- latest verified helper range was `43..62`
- Current implication:
- the Phase 5 baseline is now reproducible
- next work should target auth semantics rather than basic chain bring-up
## Phase 6.5: Concurrency and Multi-Client Test Setup ## Phase 6.5: Concurrency and Multi-Client Test Setup
1. Single-VM concurrency tests. 1. Single-VM concurrency tests.
@ -434,6 +532,14 @@ Exit criteria:
Exit criteria: Exit criteria:
- `k_proxy` can validate via wireless phone path with no client-facing API changes. - `k_proxy` can validate via wireless phone path with no client-facing API changes.
## Current Next Step
- Resolve the direct-registration blocker for `--auth-mode fido2-direct` in `k_proxy`.
- Candidate directions:
- determine whether the current card can support the required PIN/UV path for direct CTAP2 registration from `python-fido2`
- or provide a different one-time enrollment route that yields persistent real credential material for later direct assertion verification
- Keep the new regression helper as the fast check that transport, session reuse, and counter semantics still hold after each change.
## Inputs Expected During This Session ## Inputs Expected During This Session
- Exact observed behavior on reconnect attempts (USB/hidraw/probe). - Exact observed behavior on reconnect attempts (USB/hidraw/probe).

597
k_proxy_app.py
View File

@ -3,19 +3,24 @@
Minimal k_proxy service for Phase 5 bring-up. Minimal k_proxy service for Phase 5 bring-up.
Behavior: Behavior:
- Creates short-lived sessions after a card-presence check. - Creates short-lived sessions after a card-backed auth gate.
- Reuses valid sessions to access k_server protected counter endpoint. - Reuses valid sessions to access k_server protected counter endpoint.
- Supports session status and logout. - Supports enrollment, session status, and logout.
Notes: Notes:
- Session login uses `fido2_probe.py --json` command success as auth gate for now. - Default runtime still uses the legacy card-presence probe gate.
- This is a Phase 5 starter and not a final production auth design. - Experimental direct FIDO2 registration/assertion lives behind `--auth-mode fido2-direct`.
- This is still a prototype and not a final production auth design.
""" """
from __future__ import annotations from __future__ import annotations
import argparse import argparse
import base64
import http.client
import json import json
import queue
import re
import secrets import secrets
import ssl import ssl
import subprocess import subprocess
@ -29,6 +34,24 @@ from urllib.error import HTTPError, URLError
from urllib.parse import urlparse from urllib.parse import urlparse
from urllib.request import Request, urlopen from urllib.request import Request, urlopen
from fido2.client import Fido2Client, UserInteraction, verify_rp_id
from fido2.hid import CtapHidDevice
from fido2.server import Fido2Server
from fido2.webauthn import (
AttestedCredentialData,
PublicKeyCredentialCreationOptions,
PublicKeyCredentialRequestOptions,
PublicKeyCredentialRpEntity,
PublicKeyCredentialUserEntity,
UserVerificationRequirement,
)
try:
from fido2.client import ClientDataCollector, CollectedClientData
except ImportError:
ClientDataCollector = None
CollectedClientData = None
HTML = """<!doctype html> HTML = """<!doctype html>
<html lang="en"> <html lang="en">
@ -146,7 +169,7 @@ HTML = """<!doctype html>
<h1>ChromeCard Proxy Portal</h1> <h1>ChromeCard Proxy Portal</h1>
<p class="subtitle"> <p class="subtitle">
Primary browser entry point for the current prototype. Browser traffic now targets k_proxy directly. Primary browser entry point for the current prototype. Browser traffic now targets k_proxy directly.
Enrollment, login, session reuse, counter access, and logout all happen on this TLS endpoint. Enrollment, card-backed login, session reuse, counter access, and logout all happen on this TLS endpoint.
</p> </p>
</section> </section>
@ -155,9 +178,14 @@ HTML = """<!doctype html>
<h2>Enrollment</h2> <h2>Enrollment</h2>
<label for="username">Username</label> <label for="username">Username</label>
<input id="username" placeholder="alice" autocomplete="off"> <input id="username" placeholder="alice" autocomplete="off">
<label for="displayName">Display Name</label>
<input id="displayName" placeholder="Alice Example" autocomplete="off">
<div class="actions"> <div class="actions">
<button id="enrollBtn">Enroll User</button> <button id="enrollBtn">Enroll User</button>
<button id="updateBtn" class="secondary">Update User</button>
<button id="deleteBtn" class="secondary">Delete User</button>
<button id="checkBtn" class="secondary">Check Enrollment</button> <button id="checkBtn" class="secondary">Check Enrollment</button>
<button id="listBtn" class="secondary">List Users</button>
</div> </div>
<div class="status"> <div class="status">
<div>Stored username: <strong id="storedUser">none</strong></div> <div>Stored username: <strong id="storedUser">none</strong></div>
@ -185,6 +213,7 @@ HTML = """<!doctype html>
const EXP_KEY = "chromecard.proxy.expires_at"; const EXP_KEY = "chromecard.proxy.expires_at";
const logNode = document.getElementById("log"); const logNode = document.getElementById("log");
const usernameNode = document.getElementById("username"); const usernameNode = document.getElementById("username");
const displayNameNode = document.getElementById("displayName");
const storedUserNode = document.getElementById("storedUser"); const storedUserNode = document.getElementById("storedUser");
const sessionActiveNode = document.getElementById("sessionActive"); const sessionActiveNode = document.getElementById("sessionActive");
@ -226,7 +255,8 @@ HTML = """<!doctype html>
document.getElementById("enrollBtn").addEventListener("click", async () => { document.getElementById("enrollBtn").addEventListener("click", async () => {
try { try {
const username = usernameNode.value.trim(); const username = usernameNode.value.trim();
const data = await jsonRequest("POST", "/enroll/register", {username}); const display_name = displayNameNode.value.trim();
const data = await jsonRequest("POST", "/enroll/register", {username, display_name});
localStorage.setItem(USER_KEY, username); localStorage.setItem(USER_KEY, username);
syncState(); syncState();
log("Enrollment updated", data); log("Enrollment updated", data);
@ -242,11 +272,55 @@ HTML = """<!doctype html>
const data = await resp.json(); const data = await resp.json();
if (!resp.ok) throw new Error(JSON.stringify(data)); if (!resp.ok) throw new Error(JSON.stringify(data));
log("Enrollment status", data); log("Enrollment status", data);
if (data.display_name) {
displayNameNode.value = data.display_name;
}
} catch (err) { } catch (err) {
log("Enrollment status failed", {error: err.message}); log("Enrollment status failed", {error: err.message});
} }
}); });
document.getElementById("updateBtn").addEventListener("click", async () => {
try {
const username = usernameNode.value.trim() || getStoredUser();
const display_name = displayNameNode.value.trim();
const data = await jsonRequest("POST", "/enroll/update", {username, display_name});
localStorage.setItem(USER_KEY, username);
syncState();
log("Enrollment updated", data);
} catch (err) {
log("Enrollment update failed", {error: err.message});
}
});
document.getElementById("deleteBtn").addEventListener("click", async () => {
try {
const username = usernameNode.value.trim() || getStoredUser();
const data = await jsonRequest("POST", "/enroll/delete", {username});
if (getStoredUser() === username) {
localStorage.removeItem(USER_KEY);
localStorage.removeItem(TOKEN_KEY);
localStorage.removeItem(EXP_KEY);
}
displayNameNode.value = "";
syncState();
log("Enrollment deleted", data);
} catch (err) {
log("Enrollment delete failed", {error: err.message});
}
});
document.getElementById("listBtn").addEventListener("click", async () => {
try {
const resp = await fetch("/enroll/list");
const data = await resp.json();
if (!resp.ok) throw new Error(JSON.stringify(data));
log("Enrollment list", data);
} catch (err) {
log("Enrollment list failed", {error: err.message});
}
});
document.getElementById("loginBtn").addEventListener("click", async () => { document.getElementById("loginBtn").addEventListener("click", async () => {
try { try {
const username = usernameNode.value.trim() || getStoredUser(); const username = usernameNode.value.trim() || getStoredUser();
@ -307,30 +381,149 @@ class Session:
@dataclass @dataclass
class Enrollment: class Enrollment:
username: str username: str
enrolled_at: int display_name: str | None
created_at: int
updated_at: int
user_id_b64: str | None = None
credential_data_b64: str | None = None
USERNAME_PATTERN = re.compile(r"^[a-z0-9](?:[a-z0-9._-]{1,30}[a-z0-9])?$")
AUTH_MODE_PROBE = "probe"
AUTH_MODE_FIDO2_DIRECT = "fido2-direct"
def normalize_username(raw: str) -> str:
username = raw.strip().lower()
if not USERNAME_PATTERN.fullmatch(username):
raise ValueError(
"username must be 3-32 chars of lowercase letters, digits, dot, underscore, or dash"
)
return username
def normalize_display_name(raw: str | None) -> str | None:
value = (raw or "").strip()
if not value:
return None
if len(value) > 64:
raise ValueError("display_name must be 64 characters or fewer")
return value
def b64u_encode(data: bytes) -> str:
return base64.urlsafe_b64encode(data).rstrip(b"=").decode("ascii")
def b64u_decode(data: str) -> bytes:
pad = "=" * ((4 - len(data) % 4) % 4)
return base64.urlsafe_b64decode((data + pad).encode("ascii"))
def enrollment_payload(enrollment: "Enrollment", *, created: bool | None = None) -> dict[str, Any]:
payload: dict[str, Any] = {
"ok": True,
"username": enrollment.username,
"display_name": enrollment.display_name,
"created_at": enrollment.created_at,
"updated_at": enrollment.updated_at,
"has_credential": bool(enrollment.credential_data_b64),
}
if created is not None:
payload["created"] = created
return payload
if ClientDataCollector is not None and CollectedClientData is not None:
class ProxyClientDataCollector(ClientDataCollector):
def __init__(self, origin: str, rp_id: str):
if not verify_rp_id(rp_id, origin):
raise ValueError(f"origin {origin!r} is not valid for rp_id {rp_id!r}")
self.origin = origin
self.rp_id = rp_id
def collect_client_data(
self,
options: PublicKeyCredentialCreationOptions | PublicKeyCredentialRequestOptions,
) -> tuple[CollectedClientData, str]:
if isinstance(options, PublicKeyCredentialCreationOptions):
request_type = "webauthn.create"
requested_rp_id = options.rp.id
challenge = options.challenge
elif isinstance(options, PublicKeyCredentialRequestOptions):
request_type = "webauthn.get"
requested_rp_id = options.rp_id
challenge = options.challenge
else:
raise TypeError(f"unsupported options type: {type(options)!r}")
if requested_rp_id != self.rp_id:
raise ValueError(f"rp_id mismatch: expected {self.rp_id}, got {requested_rp_id}")
return CollectedClientData.create(
type=request_type,
challenge=challenge,
origin=self.origin,
), self.rp_id
else:
ProxyClientDataCollector = None
class ProxyUserInteraction(UserInteraction):
def prompt_up(self) -> None:
print("Touch the ChromeCard to continue...", flush=True)
super().prompt_up()
def request_pin(self, permissions, rp_id: str | None) -> str | None:
print("Authenticator PIN is required but not supported by this prototype.", flush=True)
return super().request_pin(permissions, rp_id)
class ProxyState: class ProxyState:
def __init__( def __init__(
self, self,
session_ttl_s: int, session_ttl_s: int,
auth_mode: str,
auth_command: str, auth_command: str,
server_base_url: str, server_base_url: str,
server_ca_file: str | None, server_ca_file: str | None,
server_max_connections: int,
proxy_token: str, proxy_token: str,
enrollment_db: Path, enrollment_db: Path,
rp_id: str,
rp_name: str,
origin: str,
): ):
self.session_ttl_s = session_ttl_s self.session_ttl_s = session_ttl_s
self.auth_mode = auth_mode
self.auth_command = auth_command self.auth_command = auth_command
self.server_base_url = server_base_url.rstrip("/") self.server_base_url = server_base_url.rstrip("/")
self.server_ca_file = server_ca_file self.server_ca_file = server_ca_file
self.proxy_token = proxy_token self.proxy_token = proxy_token
self.enrollment_db = enrollment_db self.enrollment_db = enrollment_db
self.rp_id = rp_id
self.origin = origin
self.lock = threading.Lock() self.lock = threading.Lock()
self.sessions: dict[str, Session] = {} self.sessions: dict[str, Session] = {}
self.enrollments: dict[str, Enrollment] = {} self.enrollments: dict[str, Enrollment] = {}
self.rp = PublicKeyCredentialRpEntity(id=rp_id, name=rp_name)
self.fido_server = Fido2Server(self.rp)
self.client_data_collector = (
ProxyClientDataCollector(origin=origin, rp_id=rp_id) if ProxyClientDataCollector else None
)
self.upstream = UpstreamPool(
server_base_url=self.server_base_url,
server_ca_file=self.server_ca_file,
max_connections=server_max_connections,
)
self._load_enrollments() self._load_enrollments()
def uses_direct_fido2(self) -> bool:
return self.auth_mode == AUTH_MODE_FIDO2_DIRECT
def auth_mode_label(self) -> str:
return "fido2_assertion" if self.uses_direct_fido2() else "card_presence_probe"
def _now(self) -> float: def _now(self) -> float:
return time.time() return time.time()
@ -373,39 +566,156 @@ class ProxyState:
username = str(item.get("username", "")).strip() username = str(item.get("username", "")).strip()
if not username: if not username:
continue continue
enrolled_at = int(item.get("enrolled_at", int(self._now()))) created_at = int(item.get("created_at", item.get("enrolled_at", int(self._now()))))
self.enrollments[username] = Enrollment(username=username, enrolled_at=enrolled_at) updated_at = int(item.get("updated_at", created_at))
self.enrollments[username] = Enrollment(
username=username,
display_name=normalize_display_name(item.get("display_name")),
created_at=created_at,
updated_at=updated_at,
user_id_b64=item.get("user_id_b64"),
credential_data_b64=item.get("credential_data_b64"),
)
except Exception: except Exception:
self.enrollments = {} self.enrollments = {}
def _save_enrollments_locked(self) -> None: def _save_enrollments_locked(self) -> None:
self.enrollment_db.parent.mkdir(parents=True, exist_ok=True) self.enrollment_db.parent.mkdir(parents=True, exist_ok=True)
users = [ users = [
{"username": enrollment.username, "enrolled_at": enrollment.enrolled_at} {
"username": enrollment.username,
"display_name": enrollment.display_name,
"created_at": enrollment.created_at,
"updated_at": enrollment.updated_at,
"user_id_b64": enrollment.user_id_b64,
"credential_data_b64": enrollment.credential_data_b64,
}
for enrollment in sorted(self.enrollments.values(), key=lambda item: item.username) for enrollment in sorted(self.enrollments.values(), key=lambda item: item.username)
] ]
self.enrollment_db.write_text(json.dumps({"users": users}, indent=2) + "\n") self.enrollment_db.write_text(json.dumps({"users": users}, indent=2) + "\n")
def register_enrollment(self, username: str) -> tuple[bool, Enrollment]: def _new_fido_client(self) -> Fido2Client:
username = username.strip() try:
enrolled_at = int(self._now()) device = next(CtapHidDevice.list_devices())
except StopIteration as exc:
raise RuntimeError("no CTAP HID devices found") from exc
# Newer python-fido2 builds accept a custom client-data collector, while the
# VM-side package still expects an origin string plus verifier callback.
if self.client_data_collector is not None:
return Fido2Client(device, self.client_data_collector, ProxyUserInteraction())
return Fido2Client(device, self.origin, verify=verify_rp_id, user_interaction=ProxyUserInteraction())
def _user_entity(self, username: str, display_name: str | None, user_id: bytes) -> PublicKeyCredentialUserEntity:
return PublicKeyCredentialUserEntity(
id=user_id,
name=username,
display_name=display_name or username,
)
def _register_metadata_only(self, username: str, display_name: str | None) -> Enrollment:
canonical = normalize_username(username)
pretty = normalize_display_name(display_name)
now = int(self._now())
with self.lock: with self.lock:
existing = self.enrollments.get(username) existing = self.enrollments.get(canonical)
if existing: if existing:
return False, existing raise FileExistsError("user already enrolled")
enrollment = Enrollment(username=username, enrolled_at=enrolled_at) enrollment = Enrollment(
self.enrollments[username] = enrollment username=canonical,
display_name=pretty,
created_at=now,
updated_at=now,
)
self.enrollments[canonical] = enrollment
self._save_enrollments_locked() self._save_enrollments_locked()
return True, enrollment return enrollment
def _register_direct_fido2(self, username: str, display_name: str | None) -> Enrollment:
canonical = normalize_username(username)
pretty = normalize_display_name(display_name)
now = int(self._now())
with self.lock:
existing = self.enrollments.get(canonical)
if existing and existing.credential_data_b64:
raise FileExistsError("user already enrolled")
user_id = b64u_decode(existing.user_id_b64) if existing and existing.user_id_b64 else secrets.token_bytes(32)
created_at = existing.created_at if existing else now
options, state = self.fido_server.register_begin(
self._user_entity(canonical, pretty, user_id),
user_verification=UserVerificationRequirement.DISCOURAGED,
)
try:
auth_data = self.fido_server.register_complete(
state,
self._new_fido_client().make_credential(options.public_key),
)
except Exception as exc:
raise RuntimeError(f"card registration failed: {exc}") from exc
credential_data = auth_data.credential_data
if credential_data is None:
raise RuntimeError("card registration returned no credential data")
enrollment = Enrollment(
username=canonical,
display_name=pretty,
created_at=created_at,
updated_at=now,
user_id_b64=b64u_encode(user_id),
credential_data_b64=b64u_encode(bytes(credential_data)),
)
with self.lock:
self.enrollments[canonical] = enrollment
self._save_enrollments_locked()
return enrollment
def register_enrollment(self, username: str, display_name: str | None) -> Enrollment:
if self.uses_direct_fido2():
return self._register_direct_fido2(username, display_name)
return self._register_metadata_only(username, display_name)
def update_enrollment(self, username: str, display_name: str | None) -> Enrollment:
canonical = normalize_username(username)
pretty = normalize_display_name(display_name)
now = int(self._now())
with self.lock:
existing = self.enrollments.get(canonical)
if not existing:
raise KeyError("user not enrolled")
existing.display_name = pretty
existing.updated_at = now
self._save_enrollments_locked()
return existing
def delete_enrollment(self, username: str) -> Enrollment:
canonical = normalize_username(username)
with self.lock:
existing = self.enrollments.pop(canonical, None)
if not existing:
raise KeyError("user not enrolled")
dead_tokens = [token for token, sess in self.sessions.items() if sess.username == canonical]
for token in dead_tokens:
del self.sessions[token]
self._save_enrollments_locked()
return existing
def list_enrollments(self) -> list[Enrollment]:
with self.lock:
return [self.enrollments[key] for key in sorted(self.enrollments)]
def get_enrollment(self, username: str) -> Enrollment | None: def get_enrollment(self, username: str) -> Enrollment | None:
try:
canonical = normalize_username(username)
except ValueError:
return None
with self.lock: with self.lock:
return self.enrollments.get(username.strip()) return self.enrollments.get(canonical)
def has_enrollment(self, username: str) -> bool: def has_enrollment(self, username: str) -> bool:
return self.get_enrollment(username) is not None return self.get_enrollment(username) is not None
def authenticate_with_card(self) -> tuple[bool, str]: def _authenticate_with_probe(self) -> tuple[bool, str]:
try: try:
proc = subprocess.run( proc = subprocess.run(
self.auth_command, self.auth_command,
@ -426,33 +736,107 @@ class ProxyState:
return True, "card presence check succeeded" return True, "card presence check succeeded"
def fetch_counter(self) -> tuple[int, dict[str, Any]]: def _authenticate_with_direct_fido2(self, username: str) -> tuple[bool, str]:
url = f"{self.server_base_url}/resource/counter" enrollment = self.get_enrollment(username)
req = Request(url, method="POST") if not enrollment:
req.add_header("X-Proxy-Token", self.proxy_token) return False, "user not enrolled"
req.add_header("Content-Type", "application/json") if not enrollment.credential_data_b64:
body = b"{}" return False, "user has no registered credential"
ssl_context = None
if self.server_base_url.startswith("https://"):
ssl_context = ssl.create_default_context(cafile=self.server_ca_file)
try: try:
with urlopen(req, data=body, timeout=5, context=ssl_context) as resp: credential = AttestedCredentialData(b64u_decode(enrollment.credential_data_b64))
data = json.loads(resp.read().decode("utf-8")) # Keep UV explicitly discouraged here. On the current card/library stack,
return resp.status, data # asking for stronger UV flows immediately trips PIN/UV capability errors.
except HTTPError as exc: options, state = self.fido_server.authenticate_begin(
[credential],
user_verification=UserVerificationRequirement.DISCOURAGED,
)
selection = self._new_fido_client().get_assertion(options.public_key)
assertion = selection.get_response(0)
self.fido_server.authenticate_complete(state, [credential], assertion)
except Exception as exc:
return False, f"assertion verification failed: {exc}"
return True, "assertion verified"
def authenticate_with_card(self, username: str) -> tuple[bool, str]:
if not self.uses_direct_fido2():
return self._authenticate_with_probe()
return self._authenticate_with_direct_fido2(username)
def fetch_counter(self) -> tuple[int, dict[str, Any]]:
return self.upstream.request_json(
path="/resource/counter",
headers={"X-Proxy-Token": self.proxy_token},
payload={},
)
class UpstreamPool:
def __init__(self, server_base_url: str, server_ca_file: str | None, max_connections: int = 4):
parsed = urlparse(server_base_url)
self.scheme = parsed.scheme
self.host = parsed.hostname or "127.0.0.1"
self.port = parsed.port or (443 if parsed.scheme == "https" else 80)
self.base_path = parsed.path.rstrip("/")
self.server_ca_file = server_ca_file
self.timeout = 5
self.max_connections = max_connections
self.idle: queue.LifoQueue[http.client.HTTPConnection] = queue.LifoQueue()
self.capacity = threading.BoundedSemaphore(max_connections)
def _new_connection(self) -> http.client.HTTPConnection:
if self.scheme == "https":
context = ssl.create_default_context(cafile=self.server_ca_file)
return http.client.HTTPSConnection(
self.host,
self.port,
timeout=self.timeout,
context=context,
)
return http.client.HTTPConnection(self.host, self.port, timeout=self.timeout)
def _acquire(self) -> http.client.HTTPConnection:
self.capacity.acquire()
try:
return self.idle.get_nowait()
except queue.Empty:
return self._new_connection()
def _release(self, conn: http.client.HTTPConnection | None, reusable: bool) -> None:
try:
if conn is not None and reusable:
self.idle.put(conn)
elif conn is not None:
conn.close()
finally:
self.capacity.release()
def request_json(self, path: str, headers: dict[str, str], payload: dict[str, Any]) -> tuple[int, dict[str, Any]]:
conn = self._acquire()
reusable = False
full_path = f"{self.base_path}{path}" if self.base_path else path
try:
body = json.dumps(payload).encode("utf-8")
req_headers = {"Content-Type": "application/json", **headers}
conn.request("POST", full_path, body=body, headers=req_headers)
resp = conn.getresponse()
raw = resp.read()
reusable = not resp.will_close
try: try:
data = json.loads(exc.read().decode("utf-8")) data = json.loads(raw.decode("utf-8")) if raw else {}
except Exception: except Exception:
data = {"ok": False, "error": f"server http error {exc.code}"} data = {"ok": False, "error": f"server http error {resp.status}"}
return exc.code, data return resp.status, data
except URLError as exc: except (http.client.HTTPException, OSError, ssl.SSLError) as exc:
return 502, {"ok": False, "error": f"server unavailable: {exc.reason}"} return 502, {"ok": False, "error": f"server unavailable: {exc}"}
except Exception as exc: except Exception as exc:
return 502, {"ok": False, "error": f"server call failed: {exc}"} return 502, {"ok": False, "error": f"server call failed: {exc}"}
finally:
self._release(conn, reusable)
class Handler(BaseHTTPRequestHandler): class Handler(BaseHTTPRequestHandler):
state: ProxyState state: ProxyState
protocol_version = "HTTP/1.1"
def _json(self, status: int, payload: dict[str, Any]) -> None: def _json(self, status: int, payload: dict[str, Any]) -> None:
body = json.dumps(payload).encode("utf-8") body = json.dumps(payload).encode("utf-8")
@ -477,6 +861,11 @@ class Handler(BaseHTTPRequestHandler):
return {} return {}
return json.loads(raw.decode("utf-8")) return json.loads(raw.decode("utf-8"))
def _discard_request_body(self) -> None:
length = int(self.headers.get("Content-Length", "0"))
if length > 0:
self.rfile.read(length)
def _bearer_token(self) -> str | None: def _bearer_token(self) -> str | None:
value = self.headers.get("Authorization", "") value = self.headers.get("Authorization", "")
if not value.startswith("Bearer "): if not value.startswith("Bearer "):
@ -514,6 +903,9 @@ class Handler(BaseHTTPRequestHandler):
if path.startswith("/enroll/status"): if path.startswith("/enroll/status"):
self._enroll_status() self._enroll_status()
return return
if path == "/enroll/list":
self._enroll_list()
return
self.send_error(404) self.send_error(404)
def do_POST(self) -> None: # noqa: N802 def do_POST(self) -> None: # noqa: N802
@ -524,6 +916,12 @@ class Handler(BaseHTTPRequestHandler):
if path == "/enroll/register": if path == "/enroll/register":
self._enroll_register() self._enroll_register()
return return
if path == "/enroll/update":
self._enroll_update()
return
if path == "/enroll/delete":
self._enroll_delete()
return
if path == "/session/status": if path == "/session/status":
self._session_status() self._session_status()
return return
@ -542,15 +940,16 @@ class Handler(BaseHTTPRequestHandler):
self._json(400, {"ok": False, "error": "invalid json"}) self._json(400, {"ok": False, "error": "invalid json"})
return return
username = str(data.get("username", "")).strip() try:
if not username: username = normalize_username(str(data.get("username", "")))
self._json(400, {"ok": False, "error": "username required"}) except ValueError as exc:
self._json(400, {"ok": False, "error": str(exc)})
return return
if not self.state.has_enrollment(username): if not self.state.has_enrollment(username):
self._json(403, {"ok": False, "error": "user not enrolled", "username": username}) self._json(403, {"ok": False, "error": "user not enrolled", "username": username})
return return
ok, message = self.state.authenticate_with_card() ok, message = self.state.authenticate_with_card(username)
if not ok: if not ok:
self._json(401, {"ok": False, "error": "card auth failed", "details": message}) self._json(401, {"ok": False, "error": "card auth failed", "details": message})
return return
@ -564,7 +963,7 @@ class Handler(BaseHTTPRequestHandler):
"session_token": token, "session_token": token,
"expires_at": int(expires_at), "expires_at": int(expires_at),
"ttl_seconds": self.state.session_ttl_s, "ttl_seconds": self.state.session_ttl_s,
"auth_mode": "card_presence_probe", "auth_mode": self.state.auth_mode_label(),
}, },
) )
@ -575,21 +974,57 @@ class Handler(BaseHTTPRequestHandler):
self._json(400, {"ok": False, "error": "invalid json"}) self._json(400, {"ok": False, "error": "invalid json"})
return return
username = str(data.get("username", "")).strip() try:
if not username: enrollment = self.state.register_enrollment(
self._json(400, {"ok": False, "error": "username required"}) str(data.get("username", "")),
data.get("display_name"),
)
except ValueError as exc:
self._json(400, {"ok": False, "error": str(exc)})
return
except FileExistsError:
self._json(409, {"ok": False, "error": "user already enrolled"})
return
except RuntimeError as exc:
self._json(401, {"ok": False, "error": str(exc)})
return return
created, enrollment = self.state.register_enrollment(username) self._json(200, enrollment_payload(enrollment, created=enrollment.created_at == enrollment.updated_at))
self._json(
200, def _enroll_update(self) -> None:
{ try:
"ok": True, data = self._read_json()
"username": enrollment.username, except Exception:
"enrolled_at": enrollment.enrolled_at, self._json(400, {"ok": False, "error": "invalid json"})
"created": created, return
}, try:
) enrollment = self.state.update_enrollment(
str(data.get("username", "")),
data.get("display_name"),
)
except ValueError as exc:
self._json(400, {"ok": False, "error": str(exc)})
return
except KeyError:
self._json(404, {"ok": False, "error": "user not enrolled"})
return
self._json(200, enrollment_payload(enrollment))
def _enroll_delete(self) -> None:
try:
data = self._read_json()
except Exception:
self._json(400, {"ok": False, "error": "invalid json"})
return
try:
enrollment = self.state.delete_enrollment(str(data.get("username", "")))
except ValueError as exc:
self._json(400, {"ok": False, "error": str(exc)})
return
except KeyError:
self._json(404, {"ok": False, "error": "user not enrolled"})
return
self._json(200, {"ok": True, "username": enrollment.username, "deleted": True})
def _enroll_status(self) -> None: def _enroll_status(self) -> None:
parsed = urlparse(self.path) parsed = urlparse(self.path)
@ -608,16 +1043,14 @@ class Handler(BaseHTTPRequestHandler):
if not enrollment: if not enrollment:
self._json(404, {"ok": False, "error": "user not enrolled", "username": username}) self._json(404, {"ok": False, "error": "user not enrolled", "username": username})
return return
self._json( self._json(200, enrollment_payload(enrollment))
200,
{ def _enroll_list(self) -> None:
"ok": True, users = [enrollment_payload(item) for item in self.state.list_enrollments()]
"username": enrollment.username, self._json(200, {"ok": True, "users": users})
"enrolled_at": enrollment.enrolled_at,
},
)
def _session_status(self) -> None: def _session_status(self) -> None:
self._discard_request_body()
got = self._require_session() got = self._require_session()
if not got: if not got:
return return
@ -633,6 +1066,7 @@ class Handler(BaseHTTPRequestHandler):
) )
def _session_logout(self) -> None: def _session_logout(self) -> None:
self._discard_request_body()
token = self._bearer_token() token = self._bearer_token()
if not token: if not token:
self._json(401, {"ok": False, "error": "missing bearer token"}) self._json(401, {"ok": False, "error": "missing bearer token"})
@ -641,6 +1075,7 @@ class Handler(BaseHTTPRequestHandler):
self._json(200, {"ok": True, "invalidated": removed}) self._json(200, {"ok": True, "invalidated": removed})
def _resource_counter(self) -> None: def _resource_counter(self) -> None:
self._discard_request_body()
got = self._require_session() got = self._require_session()
if not got: if not got:
return return
@ -667,10 +1102,31 @@ def parse_args() -> argparse.Namespace:
parser.add_argument("--tls-certfile", help="PEM certificate chain for HTTPS listener") parser.add_argument("--tls-certfile", help="PEM certificate chain for HTTPS listener")
parser.add_argument("--tls-keyfile", help="PEM private key for HTTPS listener") parser.add_argument("--tls-keyfile", help="PEM private key for HTTPS listener")
parser.add_argument("--session-ttl", type=int, default=300, help="Session TTL in seconds") parser.add_argument("--session-ttl", type=int, default=300, help="Session TTL in seconds")
parser.add_argument(
"--auth-mode",
choices=(AUTH_MODE_PROBE, AUTH_MODE_FIDO2_DIRECT),
default=AUTH_MODE_PROBE,
help="Session auth mode: legacy card-presence probe or experimental direct FIDO2 registration/assertion",
)
parser.add_argument( parser.add_argument(
"--auth-command", "--auth-command",
default="python3 /home/user/chromecard/fido2_probe.py --json", default="python3 /home/user/chromecard/fido2_probe.py --json",
help="Command used for session creation auth gate", help="Command used for legacy probe auth mode",
)
parser.add_argument(
"--rp-id",
default="localhost",
help="Relying party ID used for direct card-backed registration and assertion verification",
)
parser.add_argument(
"--rp-name",
default="ChromeCard Proxy",
help="Relying party name used for direct card-backed registration",
)
parser.add_argument(
"--origin",
default="https://localhost",
help="Synthetic origin used by the local FIDO2 client when talking directly to the attached card",
) )
parser.add_argument( parser.add_argument(
"--server-base-url", "--server-base-url",
@ -681,6 +1137,12 @@ def parse_args() -> argparse.Namespace:
"--server-ca-file", "--server-ca-file",
help="CA certificate used to verify HTTPS certificate presented by k_server", help="CA certificate used to verify HTTPS certificate presented by k_server",
) )
parser.add_argument(
"--server-max-connections",
type=int,
default=1,
help="Maximum concurrent pooled upstream connections from k_proxy to k_server",
)
parser.add_argument( parser.add_argument(
"--proxy-token", "--proxy-token",
default="dev-proxy-token", default="dev-proxy-token",
@ -703,11 +1165,16 @@ def main() -> int:
state = ProxyState( state = ProxyState(
session_ttl_s=args.session_ttl, session_ttl_s=args.session_ttl,
auth_mode=args.auth_mode,
auth_command=args.auth_command, auth_command=args.auth_command,
server_base_url=args.server_base_url, server_base_url=args.server_base_url,
server_ca_file=args.server_ca_file, server_ca_file=args.server_ca_file,
server_max_connections=args.server_max_connections,
proxy_token=args.proxy_token, proxy_token=args.proxy_token,
enrollment_db=Path(args.enrollment_db), enrollment_db=Path(args.enrollment_db),
rp_id=args.rp_id,
rp_name=args.rp_name,
origin=args.origin,
) )
Handler.state = state Handler.state = state
server = ThreadingHTTPServer((args.host, args.port), Handler) server = ThreadingHTTPServer((args.host, args.port), Handler)

7
k_server_app.py
View File

@ -34,6 +34,7 @@ class ServerState:
class Handler(BaseHTTPRequestHandler): class Handler(BaseHTTPRequestHandler):
state: ServerState state: ServerState
protocol_version = "HTTP/1.1"
def _json(self, status: int, payload: dict[str, Any]) -> None: def _json(self, status: int, payload: dict[str, Any]) -> None:
body = json.dumps(payload).encode("utf-8") body = json.dumps(payload).encode("utf-8")
@ -43,6 +44,11 @@ class Handler(BaseHTTPRequestHandler):
self.end_headers() self.end_headers()
self.wfile.write(body) self.wfile.write(body)
def _discard_request_body(self) -> None:
length = int(self.headers.get("Content-Length", "0"))
if length > 0:
self.rfile.read(length)
def _is_proxy_authorized(self) -> bool: def _is_proxy_authorized(self) -> bool:
return self.headers.get("X-Proxy-Token") == self.state.proxy_token return self.headers.get("X-Proxy-Token") == self.state.proxy_token
@ -65,6 +71,7 @@ class Handler(BaseHTTPRequestHandler):
if path != "/resource/counter": if path != "/resource/counter":
self.send_error(404) self.send_error(404)
return return
self._discard_request_body()
if not self._is_proxy_authorized(): if not self._is_proxy_authorized():
self._json(401, {"ok": False, "error": "unauthorized proxy"}) self._json(401, {"ok": False, "error": "unauthorized proxy"})
return return

180
phase5_chain_regression.sh Executable file
View File

@ -0,0 +1,180 @@
#!/usr/bin/env bash
set -euo pipefail
CLIENT_HOST="${CLIENT_HOST:-k_client}"
CA_FILE="${CA_FILE:-/home/user/chromecard/tls/phase2/ca.crt}"
PROXY_URL="${PROXY_URL:-https://127.0.0.1:9771}"
USERNAME="${USERNAME:-alice}"
REQUESTS="${REQUESTS:-20}"
PARALLELISM="${PARALLELISM:-8}"
CONNECT_TIMEOUT="${CONNECT_TIMEOUT:-8}"
usage() {
cat <<'EOF'
Usage: phase5_chain_regression.sh [options]
Runs the Phase 5 split-VM regression from the host by executing the client-side
flow inside k_client over SSH.
Options:
--client-host HOST SSH host alias for k_client (default: k_client)
--ca-file PATH CA bundle path inside k_client
--proxy-url URL Proxy URL visible from k_client
--username NAME Username for session login
--requests N Number of counter requests to issue
--parallelism N Number of concurrent workers
--connect-timeout SEC SSH connect timeout
-h, --help Show this help text
EOF
}
while [[ $# -gt 0 ]]; do
case "$1" in
--client-host)
CLIENT_HOST="$2"
shift 2
;;
--ca-file)
CA_FILE="$2"
shift 2
;;
--proxy-url)
PROXY_URL="$2"
shift 2
;;
--username)
USERNAME="$2"
shift 2
;;
--requests)
REQUESTS="$2"
shift 2
;;
--parallelism)
PARALLELISM="$2"
shift 2
;;
--connect-timeout)
CONNECT_TIMEOUT="$2"
shift 2
;;
-h|--help)
usage
exit 0
;;
*)
echo "unknown argument: $1" >&2
usage >&2
exit 2
;;
esac
done
ssh \
-o BatchMode=yes \
-o StrictHostKeyChecking=accept-new \
-o ConnectTimeout="${CONNECT_TIMEOUT}" \
"${CLIENT_HOST}" \
env \
CA_FILE="${CA_FILE}" \
PROXY_URL="${PROXY_URL}" \
USERNAME="${USERNAME}" \
REQUESTS="${REQUESTS}" \
PARALLELISM="${PARALLELISM}" \
python3 - <<'PY'
import concurrent.futures
import json
import os
import ssl
import sys
import urllib.error
import urllib.request
ca_file = os.environ["CA_FILE"]
proxy_url = os.environ["PROXY_URL"].rstrip("/")
username = os.environ["USERNAME"]
requests = int(os.environ["REQUESTS"])
parallelism = int(os.environ["PARALLELISM"])
if requests < 1:
raise SystemExit("REQUESTS must be >= 1")
if parallelism < 1:
raise SystemExit("PARALLELISM must be >= 1")
ctx = ssl.create_default_context(cafile=ca_file)
def post_json(path: str, payload: dict | None = None, token: str | None = None):
data = None if payload is None else json.dumps(payload).encode("utf-8")
headers = {}
if payload is not None:
headers["Content-Type"] = "application/json"
if token:
headers["Authorization"] = f"Bearer {token}"
req = urllib.request.Request(
f"{proxy_url}{path}",
data=data,
headers=headers,
method="POST",
)
try:
with urllib.request.urlopen(req, context=ctx, timeout=10) as resp:
return resp.status, json.loads(resp.read().decode("utf-8"))
except urllib.error.HTTPError as exc:
body = exc.read().decode("utf-8")
try:
payload = json.loads(body)
except json.JSONDecodeError:
payload = {"ok": False, "error": body}
return exc.code, payload
status, login = post_json("/session/login", {"username": username})
if status != 200 or "session_token" not in login:
print(json.dumps({"ok": False, "stage": "login", "status": status, "response": login}))
raise SystemExit(1)
token = login["session_token"]
values = []
def fetch_one(_: int) -> int:
status, payload = post_json("/resource/counter", {}, token=token)
if status != 200:
raise RuntimeError(json.dumps({"status": status, "response": payload}))
return int(payload["upstream"]["value"])
try:
with concurrent.futures.ThreadPoolExecutor(max_workers=parallelism) as pool:
for value in pool.map(fetch_one, range(requests)):
values.append(value)
status_resp, session = post_json("/session/status", {}, token=token)
logout_status, logout = post_json("/session/logout", {}, token=token)
invalid_status, invalid = post_json("/resource/counter", {}, token=token)
except Exception as exc:
try:
post_json("/session/logout", {}, token=token)
finally:
raise SystemExit(str(exc))
sorted_values = sorted(values)
expected = list(range(sorted_values[0], sorted_values[-1] + 1)) if sorted_values else []
summary = {
"ok": True,
"username": username,
"proxy_url": proxy_url,
"requests": requests,
"parallelism": parallelism,
"unique": len(set(values)) == len(values),
"gap_free": sorted_values == expected,
"min": min(sorted_values) if sorted_values else None,
"max": max(sorted_values) if sorted_values else None,
"values": sorted_values,
"login": login,
"session_status": {"status": status_resp, "response": session},
"logout": {"status": logout_status, "response": logout},
"post_logout": {"status": invalid_status, "response": invalid},
}
print(json.dumps(summary, indent=2, sort_keys=True))
if not summary["unique"] or not summary["gap_free"] or logout_status != 200 or invalid_status != 401:
raise SystemExit(1)
PY

188
phase65_concurrency_probe.py Normal file
View File

@ -0,0 +1,188 @@
#!/usr/bin/env python3
"""
Phase 6.5 concurrency probe for the direct browser-to-k_proxy path.
What it does:
- Creates a small batch of enrolled users.
- Logs each user in through k_proxy over TLS.
- Fires protected counter requests in parallel using the returned bearer tokens.
- Verifies that all calls succeed and that returned counter values are unique and contiguous.
"""
from __future__ import annotations
import argparse
import json
import ssl
import sys
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass
from typing import Any
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen
@dataclass
class Session:
username: str
token: str
def request_json(
base_url: str,
path: str,
*,
method: str = "GET",
payload: dict[str, Any] | None = None,
token: str | None = None,
cafile: str | None = None,
timeout: int = 10,
) -> tuple[int, dict[str, Any]]:
req = Request(f"{base_url.rstrip('/')}{path}", method=method)
req.add_header("Content-Type", "application/json")
if token:
req.add_header("Authorization", f"Bearer {token}")
data = None if payload is None else json.dumps(payload).encode("utf-8")
context = ssl.create_default_context(cafile=cafile) if base_url.startswith("https://") else None
try:
with urlopen(req, data=data, timeout=timeout, context=context) as resp:
return resp.status, json.loads(resp.read().decode("utf-8"))
except HTTPError as exc:
try:
return exc.code, json.loads(exc.read().decode("utf-8"))
except Exception:
return exc.code, {"ok": False, "error": f"http error {exc.code}"}
except URLError as exc:
return 502, {"ok": False, "error": f"url error: {exc.reason}"}
except Exception as exc:
return 502, {"ok": False, "error": f"request failed: {exc}"}
def enroll_user(base_url: str, cafile: str, username: str, display_name: str) -> None:
status, data = request_json(
base_url,
"/enroll/register",
method="POST",
payload={"username": username, "display_name": display_name},
cafile=cafile,
)
if status == 200:
return
if status == 409 and data.get("error") == "user already enrolled":
return
raise RuntimeError(f"enroll failed for {username}: status={status} data={data}")
def login_user(base_url: str, cafile: str, username: str) -> Session:
status, data = request_json(
base_url,
"/session/login",
method="POST",
payload={"username": username},
cafile=cafile,
)
if status != 200 or not data.get("session_token"):
raise RuntimeError(f"login failed for {username}: status={status} data={data}")
return Session(username=username, token=data["session_token"])
def counter_call(base_url: str, cafile: str, session: Session, call_id: int) -> dict[str, Any]:
started = time.time()
status, data = request_json(
base_url,
"/resource/counter",
method="POST",
payload={},
token=session.token,
cafile=cafile,
)
finished = time.time()
return {
"call_id": call_id,
"username": session.username,
"status": status,
"data": data,
"latency_ms": int((finished - started) * 1000),
}
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description="Run Phase 6.5 concurrency probe against k_proxy")
parser.add_argument("--base-url", default="https://127.0.0.1:9771")
parser.add_argument("--ca-file", required=True)
parser.add_argument("--users", type=int, default=3)
parser.add_argument("--requests-per-user", type=int, default=4)
parser.add_argument("--username-prefix", default="phase65")
parser.add_argument(
"--max-workers",
type=int,
help="Maximum number of in-flight protected calls; defaults to total requests",
)
return parser.parse_args()
def main() -> int:
args = parse_args()
sessions: list[Session] = []
for idx in range(args.users):
username = f"{args.username_prefix}_{idx}"
enroll_user(args.base_url, args.ca_file, username, f"Phase65 User {idx}")
sessions.append(login_user(args.base_url, args.ca_file, username))
jobs: list[tuple[Session, int]] = []
call_id = 0
for session in sessions:
for _ in range(args.requests_per_user):
jobs.append((session, call_id))
call_id += 1
results: list[dict[str, Any]] = []
max_workers = args.max_workers or len(jobs)
with ThreadPoolExecutor(max_workers=max_workers) as executor:
future_map = {
executor.submit(counter_call, args.base_url, args.ca_file, session, job_id): (session.username, job_id)
for session, job_id in jobs
}
for future in as_completed(future_map):
username, job_id = future_map[future]
try:
results.append(future.result())
except Exception as exc:
results.append(
{
"call_id": job_id,
"username": username,
"status": 599,
"data": {"ok": False, "error": str(exc)},
"latency_ms": -1,
}
)
results.sort(key=lambda item: item["call_id"])
ok_results = [item for item in results if item["status"] == 200 and item["data"].get("ok")]
values = [item["data"]["upstream"]["value"] for item in ok_results]
values_sorted = sorted(values)
contiguous = bool(values_sorted) and values_sorted == list(range(values_sorted[0], values_sorted[0] + len(values_sorted)))
summary = {
"ok": len(ok_results) == len(results) and len(set(values)) == len(values) and contiguous,
"users": args.users,
"requests_per_user": args.requests_per_user,
"total_requests": len(results),
"max_workers": max_workers,
"successful_requests": len(ok_results),
"unique_counter_values": len(set(values)),
"counter_min": min(values_sorted) if values_sorted else None,
"counter_max": max(values_sorted) if values_sorted else None,
"counter_contiguous": contiguous,
"max_latency_ms": max((item["latency_ms"] for item in results), default=None),
"results": results,
}
print(json.dumps(summary, indent=2))
return 0 if summary["ok"] else 1
if __name__ == "__main__":
raise SystemExit(main())

311
raw_ctap_probe.py Normal file
View File

@ -0,0 +1,311 @@
#!/usr/bin/env python3
"""
Low-level CTAP2 probe for ChromeCard host debugging.
This bypasses the higher-level Fido2Client/WebAuthn helpers so we can inspect
raw makeCredential/getAssertion behavior, keepalive callbacks, and transport
errors on the host stack.
"""
from __future__ import annotations
import argparse
import hashlib
import json
import secrets
import sys
import time
import traceback
from typing import Any
try:
from fido2.ctap import CtapError
from fido2.ctap2 import Ctap2
from fido2.hid import CtapHidDevice
except Exception as exc:
print("Missing dependency: python-fido2", file=sys.stderr)
print("Install with: python3 -m pip install fido2", file=sys.stderr)
print(f"Import error: {exc}", file=sys.stderr)
sys.exit(2)
def _json_default(value: Any) -> Any:
if isinstance(value, bytes):
return value.hex()
if isinstance(value, set):
return sorted(value)
if hasattr(value, "items"):
return dict(value.items())
return str(value)
def _now() -> str:
return time.strftime("%Y-%m-%dT%H:%M:%S", time.localtime())
def log(message: str) -> None:
print(f"[{_now()}] {message}", file=sys.stderr, flush=True)
def list_devices() -> list[CtapHidDevice]:
return list(CtapHidDevice.list_devices())
def describe_device(dev: CtapHidDevice) -> dict[str, Any]:
desc = getattr(dev, "descriptor", None)
return {
"product_name": getattr(desc, "product_name", None),
"manufacturer": getattr(desc, "manufacturer_string", None),
"vendor_id": getattr(desc, "vid", None),
"product_id": getattr(desc, "pid", None),
"path": getattr(desc, "path", None),
}
def get_ctap2(dev: CtapHidDevice) -> Ctap2:
return Ctap2(dev)
def print_json(payload: dict[str, Any]) -> None:
print(json.dumps(payload, indent=2, default=_json_default))
def keepalive_logger(status: int) -> None:
log(f"keepalive status={status}")
def _coerce_hex_bytes(value: str | None, label: str) -> bytes | None:
if value is None:
return None
raw = value.strip().lower()
if raw.startswith("0x"):
raw = raw[2:]
try:
return bytes.fromhex(raw)
except ValueError as exc:
raise SystemExit(f"invalid hex for {label}: {value}") from exc
def _client_data_hash(label: str) -> bytes:
return hashlib.sha256(label.encode("utf-8")).digest()
def _key_params() -> list[dict[str, Any]]:
return [
{"type": "public-key", "alg": -7},
{"type": "public-key", "alg": -257},
]
def do_info(ctap2: Ctap2, device_meta: dict[str, Any]) -> int:
info = ctap2.get_info()
print_json({"device": device_meta, "ctap2_info": info})
return 0
def do_make_credential(ctap2: Ctap2, args: argparse.Namespace, device_meta: dict[str, Any]) -> int:
rp = {"id": args.rp_id, "name": args.rp_name or args.rp_id}
user_id = args.user_id.encode("utf-8")
user = {
"id": user_id,
"name": args.user_name,
"displayName": args.user_display_name or args.user_name,
}
client_data_hash = _client_data_hash(f"chromecard-make-credential:{args.rp_id}:{args.user_name}")
options = {"rk": args.resident_key, "uv": args.user_verification}
log(
"starting makeCredential "
f"rp_id={args.rp_id} user={args.user_name} rk={options['rk']} uv={options['uv']}"
)
try:
response = ctap2.make_credential(
client_data_hash=client_data_hash,
rp=rp,
user=user,
key_params=_key_params(),
options=options,
on_keepalive=keepalive_logger,
)
except CtapError as exc:
print_json(
{
"operation": "makeCredential",
"device": device_meta,
"rp": rp,
"user": user,
"options": options,
"error_type": "CtapError",
"error_code": getattr(exc, "code", None),
"error_name": str(getattr(exc, "code", None)),
"message": str(exc),
}
)
return 1
except Exception as exc:
print_json(
{
"operation": "makeCredential",
"device": device_meta,
"rp": rp,
"user": user,
"options": options,
"error_type": type(exc).__name__,
"message": str(exc),
"traceback": traceback.format_exc(),
}
)
return 1
auth_data = getattr(response, "auth_data", None)
credential_data = getattr(auth_data, "credential_data", None)
print_json(
{
"operation": "makeCredential",
"device": device_meta,
"rp": rp,
"user": user,
"options": options,
"fmt": getattr(response, "fmt", None),
"auth_data": auth_data,
"credential_id_hex": getattr(credential_data, "credential_id", b"").hex()
if credential_data is not None
else None,
"credential_data_hex": bytes(credential_data).hex() if credential_data is not None else None,
"att_stmt": getattr(response, "att_stmt", None),
}
)
return 0
def do_get_assertion(ctap2: Ctap2, args: argparse.Namespace, device_meta: dict[str, Any]) -> int:
allow_credential = _coerce_hex_bytes(args.allow_credential_id, "allow-credential-id")
allow_list = [{"type": "public-key", "id": allow_credential}] if allow_credential else None
client_data_hash = _client_data_hash(f"chromecard-get-assertion:{args.rp_id}")
options = {"up": True, "uv": args.user_verification}
log(
"starting getAssertion "
f"rp_id={args.rp_id} allow_list={1 if allow_list else 0} uv={options['uv']}"
)
try:
response = ctap2.get_assertion(
rp_id=args.rp_id,
client_data_hash=client_data_hash,
allow_list=allow_list,
options=options,
on_keepalive=keepalive_logger,
)
except CtapError as exc:
print_json(
{
"operation": "getAssertion",
"device": device_meta,
"rp_id": args.rp_id,
"allow_list": allow_list,
"options": options,
"error_type": "CtapError",
"error_code": getattr(exc, "code", None),
"error_name": str(getattr(exc, "code", None)),
"message": str(exc),
}
)
return 1
except Exception as exc:
print_json(
{
"operation": "getAssertion",
"device": device_meta,
"rp_id": args.rp_id,
"allow_list": allow_list,
"options": options,
"error_type": type(exc).__name__,
"message": str(exc),
"traceback": traceback.format_exc(),
}
)
return 1
assertions: list[dict[str, Any]] = []
for item in getattr(response, "assertions", []) or []:
assertions.append(
{
"credential": getattr(item, "credential", None),
"auth_data": getattr(item, "auth_data", None),
"signature": getattr(item, "signature", None),
"user": getattr(item, "user", None),
"number_of_credentials": getattr(item, "number_of_credentials", None),
}
)
print_json(
{
"operation": "getAssertion",
"device": device_meta,
"rp_id": args.rp_id,
"allow_list": allow_list,
"options": options,
"assertions": assertions,
}
)
return 0
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(description="Low-level CTAP2 host probe")
parser.add_argument("--index", type=int, default=0, help="Device index from --list output")
subparsers = parser.add_subparsers(dest="command", required=True)
subparsers.add_parser("list", help="List CTAP HID devices")
subparsers.add_parser("info", help="Fetch CTAP2 getInfo")
make_credential = subparsers.add_parser("make-credential", help="Run raw CTAP2 makeCredential")
make_credential.add_argument("--rp-id", default="localhost")
make_credential.add_argument("--rp-name", default="ChromeCard Local Probe")
make_credential.add_argument("--user-name", default="probe-user")
make_credential.add_argument("--user-display-name", default="Probe User")
make_credential.add_argument("--user-id", default=secrets.token_hex(16))
make_credential.add_argument("--resident-key", action="store_true")
make_credential.add_argument("--user-verification", action="store_true")
get_assertion = subparsers.add_parser("get-assertion", help="Run raw CTAP2 getAssertion")
get_assertion.add_argument("--rp-id", default="localhost")
get_assertion.add_argument("--allow-credential-id", help="Credential id as hex")
get_assertion.add_argument("--user-verification", action="store_true")
return parser
def main() -> int:
parser = build_parser()
args = parser.parse_args()
devs = list_devices()
if args.command == "list":
print_json(
{
"devices": [describe_device(dev) for dev in devs],
}
)
return 0 if devs else 1
if not devs:
print("No CTAP HID devices found.", file=sys.stderr)
return 1
if args.index < 0 or args.index >= len(devs):
print(f"Invalid --index {args.index}; found {len(devs)} device(s).", file=sys.stderr)
return 2
dev = devs[args.index]
device_meta = describe_device(dev)
ctap2 = get_ctap2(dev)
if args.command == "info":
return do_info(ctap2, device_meta)
if args.command == "make-credential":
return do_make_credential(ctap2, args, device_meta)
if args.command == "get-assertion":
return do_get_assertion(ctap2, args, device_meta)
parser.error(f"unsupported command: {args.command}")
return 2
if __name__ == "__main__":
raise SystemExit(main())