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

Switch token binding from per-request URL+method to domain-level host+nonce 

Challenge is now SHA256(host|nonce) instead of SHA256(url|method|nonce).
A single card interaction authorises access to any path and method on the
gated domain, which is the intended granularity. Tests updated accordingly:
path/method rejection cases replaced with domain-level and tampered-host cases.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Morten V. Christiansen 2026-05-09 23:52:48 +02:00
parent 139698cab5
commit 4b719a0846
5 changed files with 72 additions and 101 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
k_phone/lib/filter_proxy.dart
View File

@ -406,7 +406,7 @@ class FilterProxy {
) async {
String token;
try {
token = await _getAuthToken(uri, method);
token = await _getAuthToken(uri);
} catch (_) {
_deny(client, sub, 407, 'Proxy Authentication Required');
return;
@ -450,14 +450,13 @@ class FilterProxy {
await _forwardHttpRequest(client, sub, upstream, out.toString(), remainder, contentLength);
}
// Calls POST /auth/get-token on Component 2 with per-request binding and
// Calls POST /auth/get-token on Component 2 with domain-level binding and
// returns the bearer token (a self-contained FIDO2 assertion bundle).
// Throws if card is unavailable or Component 2 is unreachable.
Future<String> _getAuthToken(Uri uri, String method) async {
Future<String> _getAuthToken(Uri uri) async {
final nonce = _secureHex(16);
final payload = utf8.encode(jsonEncode({
'url': uri.toString(),
'method': method,
'host': uri.host,
'nonce': nonce,
}));
final httpClient = HttpClient()

20
k_phone/lib/proxy_service.dart
View File

@ -448,11 +448,11 @@ class _ProxyServer {
}
// -------------------------------------------------------------------------
// Auth token endpoint (v2 architecture — per-request token binding)
// Auth token endpoint (v2 architecture — domain-level token binding)
//
// Component 1 (filter_proxy) and Component 3 (Go binary) call this with
// {url, method, nonce} for each gated request. A fresh FIDO2 assertion is
// produced with challenge = SHA256(url|method|nonce). The self-contained
// {host, nonce} for each gated request. A fresh FIDO2 assertion is
// produced with challenge = SHA256(host|nonce). The self-contained
// assertion bundle is returned as a base64url Bearer token the server can
// verify without calling back to this service.
// -------------------------------------------------------------------------
@ -461,12 +461,11 @@ class _ProxyServer {
final body = await _readJson(req);
if (body == null) return;
final url = body['url'] as String? ?? '';
final method = body['method'] as String? ?? '';
final host = body['host'] as String? ?? '';
final nonce = body['nonce'] as String? ?? '';
if (url.isEmpty || method.isEmpty || nonce.isEmpty) {
await _send(req.response, 400, {'ok': false, 'error': 'url, method, nonce required'});
if (host.isEmpty || nonce.isEmpty) {
await _send(req.response, 400, {'ok': false, 'error': 'host, nonce required'});
return;
}
@ -487,9 +486,9 @@ class _ProxyServer {
return;
}
// Challenge = SHA256(url | "|" | method | "|" | nonce)
// Challenge = SHA256(host | "|" | nonce)
final challenge = Uint8List.fromList(
sha256.convert(utf8.encode('$url|$method|$nonce')).bytes,
sha256.convert(utf8.encode('$host|$nonce')).bytes,
);
GetAssertionResult assertionResult;
@ -503,8 +502,7 @@ class _ProxyServer {
// Self-contained bundle the server can verify without calling back to the phone.
final bundleJson = jsonEncode({
'v': 1,
'url': url,
'method': method,
'host': host,
'nonce': nonce,
'authData': base64Url.encode(assertionResult.authData).replaceAll('=', ''),
'sig': base64Url.encode(assertionResult.signature).replaceAll('=', ''),

14
k_phone/test/filter_proxy_test.dart
View File

@ -250,7 +250,7 @@ void main() {
expect(req.uri.path, '/auth/get-token');
});
test('gated host: /auth/get-token body carries url, method, nonce', () async {
test('gated host: /auth/get-token body carries host, nonce', () async {
await _round(
proxy.port,
'GET http://127.0.0.1:${endpoint.port}/api?x=1 HTTP/1.1\r\n'
@ -259,13 +259,14 @@ void main() {
final (:req, :rawBody) = await comp2TokenReq.future.timeout(_kTimeout);
expect(req.uri.path, '/auth/get-token');
final body = jsonDecode(rawBody) as Map<String, dynamic>;
expect(body['url'], contains('/api?x=1'));
expect(body['method'], 'GET');
expect(body['host'], '127.0.0.1');
expect(body.containsKey('url'), isFalse);
expect(body.containsKey('method'), isFalse);
expect(body['nonce'], isA<String>());
expect((body['nonce'] as String).length, greaterThan(8));
});
test('gated POST: /auth/get-token body carries method POST', () async {
test('gated POST: /auth/get-token body carries host, nonce (no method)', () async {
const postBody = '{"key":"val"}';
await _round(
proxy.port,
@ -278,8 +279,9 @@ void main() {
final (:req, :rawBody) = await comp2TokenReq.future.timeout(_kTimeout);
expect(req.uri.path, '/auth/get-token');
final body = jsonDecode(rawBody) as Map<String, dynamic>;
expect(body['method'], 'POST');
expect(body['url'], contains('/submit'));
expect(body['host'], '127.0.0.1');
expect(body.containsKey('method'), isFalse);
expect(body['nonce'], isA<String>());
});
test('gated host: request goes directly to endpoint with Bearer token', () async {

27
k_server_app.py
View File

@ -26,13 +26,12 @@ def _b64u_decode(s: str) -> bytes:
return base64.urlsafe_b64decode(padded)
def _verify_assertion_token(token: str, request_path: str, request_method: str) -> bool:
"""Verify a base64url-encoded FIDO2 per-request assertion bundle.
def _verify_assertion_token(token: str) -> bool:
"""Verify a base64url-encoded FIDO2 domain-level assertion bundle.
Bundle fields (JSON, then base64url-encoded):
v version (1)
url full URL used to derive the challenge
method HTTP method used to derive the challenge
host hostname used to derive the challenge
nonce random hex nonce used to derive the challenge
authData base64url authenticator data
sig base64url ECDSA signature
@ -53,19 +52,11 @@ def _verify_assertion_token(token: str, request_path: str, request_method: str)
bundle = json.loads(_b64u_decode(token).decode("utf-8"))
# Path and method must match the actual request.
bundle_path = urlparse(bundle["url"]).path
if bundle_path != request_path:
return False
if bundle["method"].upper() != request_method.upper():
return False
url = bundle["url"]
method = bundle["method"]
host = bundle["host"]
nonce = bundle["nonce"]
# Verify challenge claim: challenge == b64u(SHA256(url|method|nonce))
binding = f"{url}|{method}|{nonce}".encode()
# Verify challenge claim: challenge == b64u(SHA256(host|nonce))
binding = f"{host}|{nonce}".encode()
expected_challenge = base64.urlsafe_b64encode(hashlib.sha256(binding).digest()).rstrip(b"=").decode()
cdj_bytes = _b64u_decode(bundle["cdj"])
@ -139,11 +130,7 @@ class Handler(BaseHTTPRequestHandler):
return True
auth = self.headers.get("Authorization", "")
if auth.startswith("Bearer "):
return _verify_assertion_token(
auth[7:].strip(),
request_path=urlparse(self.path).path,
request_method=self.command,
)
return _verify_assertion_token(auth[7:].strip())
return False
def do_GET(self) -> None: # noqa: N802

103
tests/test_k_server.py
View File

@ -73,7 +73,7 @@ def _cose_es256(x: bytes, y: bytes) -> bytes:
)
def _make_bundle(url: str, method: str, nonce: str) -> tuple[str, object]:
def _make_bundle(host: str, nonce: str) -> tuple[str, object]:
"""Return (base64url-token, private_key) for a fresh P-256 assertion.
Mirrors exactly what proxy_service.dart's _handleAuthGetToken produces.
@ -90,8 +90,8 @@ def _make_bundle(url: str, method: str, nonce: str) -> tuple[str, object]:
cred_id = os.urandom(16)
cred_data = aaguid + struct.pack(">H", len(cred_id)) + cred_id + cose_key
# Challenge = SHA256(url|method|nonce) — same as proxy_service.dart
challenge_b64u = _b64u_encode(hashlib.sha256(f"{url}|{method}|{nonce}".encode()).digest())
# Challenge = SHA256(host|nonce) — same as proxy_service.dart
challenge_b64u = _b64u_encode(hashlib.sha256(f"{host}|{nonce}".encode()).digest())
cdj = json.dumps(
{
@ -112,8 +112,7 @@ def _make_bundle(url: str, method: str, nonce: str) -> tuple[str, object]:
bundle = {
"v": 1,
"url": url,
"method": method,
"host": host,
"nonce": nonce,
"authData": _b64u_encode(auth_data),
"sig": _b64u_encode(sig),
@ -137,33 +136,30 @@ def _tamper(token: str, key: str, transform) -> str:
@unittest.skipUnless(HAS_CRYPTO, "cbor2 / cryptography not installed")
class TestVerifyAssertionToken(unittest.TestCase):
def setUp(self):
self.url = "https://127.0.0.1:8780/resource/counter"
self.method = "POST"
self.host = "127.0.0.1"
self.nonce = "deadbeef01234567"
self.token, _ = _make_bundle(self.url, self.method, self.nonce)
self.token, _ = _make_bundle(self.host, self.nonce)
def _check(self, token=None, path="/resource/counter", method="POST") -> bool:
def _check(self, token=None) -> bool:
return k_server_app._verify_assertion_token(
self.token if token is None else token, request_path=path, request_method=method
self.token if token is None else token
)
def test_valid_token_accepted(self):
self.assertTrue(self._check())
def test_wrong_path_rejected(self):
self.assertFalse(self._check(path="/resource/other"))
def test_wrong_method_rejected(self):
self.assertFalse(self._check(method="GET"))
def test_method_comparison_case_insensitive(self):
self.assertTrue(self._check(method="post"))
self.assertTrue(self._check(method="POST"))
def test_token_valid_for_any_path_on_host(self):
# Domain-level binding: the same token covers all paths on the host.
self.assertTrue(self._check())
def test_tampered_nonce_invalidates_challenge(self):
tampered = _tamper(self.token, "nonce", lambda _: "tampered00000000")
self.assertFalse(self._check(tampered))
def test_tampered_host_invalidates_challenge(self):
tampered = _tamper(self.token, "host", lambda _: "attacker.com")
self.assertFalse(self._check(tampered))
def test_tampered_signature_rejected(self):
def flip_last_byte(b64: str) -> str:
raw = bytearray(_b64u_decode(b64))
@ -188,10 +184,6 @@ class TestVerifyAssertionToken(unittest.TestCase):
tampered = _tamper(self.token, "cred", swap_key)
self.assertFalse(self._check(tampered))
def test_cross_resource_replay_rejected(self):
# Token bound to /resource/counter must not pass for /resource/admin.
self.assertFalse(self._check(path="/resource/admin"))
def test_malformed_token_returns_false(self):
self.assertFalse(self._check(token="!!!not-base64!!!"))
@ -229,8 +221,7 @@ class TestVerifyAssertionTokenRoundTrip(unittest.TestCase):
def _register_and_assert(
self,
emulator: "CardEmulator",
url: str,
method: str,
host: str,
nonce: str,
) -> str:
"""Return a token string after one register + one assertion."""
@ -248,9 +239,9 @@ class TestVerifyAssertionTokenRoundTrip(unittest.TestCase):
cred_id_len = (cred_data[16] << 8) | cred_data[17]
cred_id = cred_data[18:18 + cred_id_len]
# 2. Assert with bound challenge — mirrors _handleAuthGetToken in proxy_service.dart
# 2. Assert with domain-level challenge — mirrors _handleAuthGetToken in proxy_service.dart
challenge_b64u = _b64u_encode(
hashlib.sha256(f"{url}|{method}|{nonce}".encode()).digest()
hashlib.sha256(f"{host}|{nonce}".encode()).digest()
)
cdj = json.dumps(
{
@ -271,8 +262,7 @@ class TestVerifyAssertionTokenRoundTrip(unittest.TestCase):
# 3. Encode bundle — same as proxy_service.dart _handleAuthGetToken
bundle = {
"v": 1,
"url": url,
"method": method,
"host": host,
"nonce": nonce,
"authData": _b64u_encode(bytes(assertion.auth_data)),
"sig": _b64u_encode(bytes(assertion.signature)),
@ -284,53 +274,48 @@ class TestVerifyAssertionTokenRoundTrip(unittest.TestCase):
def test_roundtrip_accepted(self):
emulator = CardEmulator()
token = self._register_and_assert(
emulator, "https://example.com/api/data", "GET", "cafebabe12345678"
)
token = self._register_and_assert(emulator, "example.com", "cafebabe12345678")
self.assertTrue(
k_server_app._verify_assertion_token(token, "/api/data", "GET"),
k_server_app._verify_assertion_token(token),
"valid round-trip token must be accepted",
)
def test_roundtrip_wrong_path_rejected(self):
def test_roundtrip_valid_for_any_path(self):
"""Domain-level token: accepted regardless of which path is requested."""
emulator = CardEmulator()
token = self._register_and_assert(
emulator, "https://example.com/api/data", "GET", "aabbccdd11223344"
)
self.assertFalse(
k_server_app._verify_assertion_token(token, "/api/other", "GET"),
"token for /api/data must not pass for /api/other",
token = self._register_and_assert(emulator, "example.com", "aabbccdd11223344")
self.assertTrue(
k_server_app._verify_assertion_token(token),
"domain-level token must be accepted for any path on the host",
)
def test_roundtrip_wrong_method_rejected(self):
emulator = CardEmulator()
token = self._register_and_assert(
emulator, "https://example.com/submit", "POST", "1122334455667788"
)
self.assertFalse(
k_server_app._verify_assertion_token(token, "/submit", "GET"),
"token for POST must not pass for GET",
)
def test_roundtrip_same_token_twice_rejected_after_nonce_tamper(self):
def test_roundtrip_same_token_tampered_nonce_rejected(self):
"""Changing the nonce in a real assertion bundle breaks verification."""
emulator = CardEmulator()
token = self._register_and_assert(
emulator, "https://example.com/resource/counter", "POST", "original00000000"
)
token = self._register_and_assert(emulator, "example.com", "original00000000")
tampered = _tamper(token, "nonce", lambda _: "tampered11111111")
self.assertFalse(
k_server_app._verify_assertion_token(tampered, "/resource/counter", "POST"),
k_server_app._verify_assertion_token(tampered),
"tampered nonce must break challenge verification",
)
def test_roundtrip_tampered_host_rejected(self):
"""Changing the host in the bundle breaks challenge verification."""
emulator = CardEmulator()
token = self._register_and_assert(emulator, "example.com", "deadbeef00112233")
tampered = _tamper(token, "host", lambda _: "attacker.com")
self.assertFalse(
k_server_app._verify_assertion_token(tampered),
"tampered host must break challenge verification",
)
def test_roundtrip_replayed_for_different_user_rejected(self):
"""Two users register separate credentials; each token is only valid for its own key."""
em_a = CardEmulator()
em_b = CardEmulator()
url, method, nonce = "https://example.com/protected", "GET", "00112233aabbccdd"
token_a = self._register_and_assert(em_a, url, method, nonce)
token_b = self._register_and_assert(em_b, url, method, nonce)
host, nonce = "example.com", "00112233aabbccdd"
token_a = self._register_and_assert(em_a, host, nonce)
token_b = self._register_and_assert(em_b, host, nonce)
# token_a's signature was made with em_a's key — must not verify with em_b's public key.
# Tamper: swap cred (public key) from token_b into token_a's bundle.
@ -340,7 +325,7 @@ class TestVerifyAssertionTokenRoundTrip(unittest.TestCase):
cross = _b64u_encode(json.dumps(bundle_a, separators=(",", ":")).encode())
self.assertFalse(
k_server_app._verify_assertion_token(cross, "/protected", "GET"),
k_server_app._verify_assertion_token(cross),
"cross-user key swap must fail verification",
)