Take the portal's JWKS from the tunnel and verify by kid

The welcome frame's keys replace the set pinned at enrolment, but only
over wss:// or loopback, the rule enrolment applies to the portal
address. Portal tokens are verified against the key their kid names, so
the hub can publish a new key beside the old one. A welcome without keys,
from an older hub, leaves the pinned set alone.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01WgBfkm4rCmtd1T4vHWjEPx
This commit is contained in:
2026-10-15 10:31:40 +02:00
co-authored by Claude Opus 5.5
parent 3a7c10b8a9
commit ffa044724b
5 changed files with 83 additions and 12 deletions
+38
View File
@@ -772,3 +772,41 @@ def test_admin_lists_and_revokes_portal_members(
)
db.delete(arrived)
db.commit()
def test_the_portal_keys_follow_the_tunnel(
enrolled: User, # noqa: ARG001 (fixture installs the enrolment)
portal_key: rsa.RSAPrivateKey,
) -> None:
"""A rotated keypair arrives on attach, and either key verifies by kid."""
import json
from cryptography.hazmat.primitives import serialization
from fluksio.cloud.connector import CloudConnector
pinned = cloud_config.load()
assert pinned is not None
successor = rsa.generate_private_key(public_exponent=65537, key_size=2048)
jwk = json.loads(jwt.algorithms.RSAAlgorithm.to_jwk(successor.public_key()))
jwk.update({"use": "sig", "alg": "RS256", "kid": "next"})
rotated = {"keys": [*jwks(portal_key)["keys"], jwk]}
claims = jwt.decode(portal_token(portal_key), options={"verify_signature": False})
pem = successor.private_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PrivateFormat.PKCS8,
encryption_algorithm=serialization.NoEncryption(),
)
signed_by_next = jwt.encode(claims, pem, algorithm="RS256", headers={"kid": "next"})
# Over cleartext to another machine, a key is not worth taking.
CloudConnector._adopt_jwks(rotated)
assert cloud_config.load().jwks == pinned.jwks
with pytest.raises(jwt.exceptions.InvalidTokenError):
decode_token(signed_by_next)
cloud_config.save(replace(pinned, ws_url="wss://hub.example.test/attach"))
CloudConnector._adopt_jwks({"keys": []})
CloudConnector._adopt_jwks(rotated)
assert decode_token(portal_token(portal_key))["portal_sub"] == "portal-user-1"
assert decode_token(signed_by_next)["portal_sub"] == "portal-user-1"