feat: add device public key and fingerprint for cross-layer binding

alembic/versions/a1b2c3d4e5f7_add_public_key_to_devices.py

@@ -0,0 +1,41 @@
+"""add public_key and device_public_key_fingerprint to devices
+
+Revision ID: a1b2c3d4e5f7
+Revises: f6a7b8c9d0e1
+Create Date: 2026-02-16 10:00:00.000000
+
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "a1b2c3d4e5f7"
+down_revision: str | None = "f6a7b8c9d0e1"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    # Pre-launch: delete existing devices (they lack a public key and
+    # cannot produce valid sidecars with cross-layer binding).
+    # Captures reference devices via FK, so delete them first.
+    op.execute("DELETE FROM captures")
+    op.execute("DELETE FROM devices")
+
+    op.add_column(
+        "devices",
+        sa.Column("public_key", sa.String(120), nullable=False),
+    )
+    op.add_column(
+        "devices",
+        sa.Column("device_public_key_fingerprint", sa.String(64), nullable=False),
+    )
+
+
+def downgrade() -> None:
+    op.drop_column("devices", "device_public_key_fingerprint")
+    op.drop_column("devices", "public_key")

app/api/routes/capture.py

@@ -41,6 +41,7 @@ async def create_session(
         device.id,
         attestation_method=device.attestation_method,
         app_id=device.attested_app_id,
+        device_public_key_fingerprint=device.device_public_key_fingerprint,
     )
 
     return CaptureSessionResponse(
@@ -85,6 +86,7 @@ async def create_trust_token(
         session_data.device_id,
         method=cast("AttestationMethod", session_data.attestation_method or "sandbox"),
         app_id=session_data.app_id,
+        device_public_key_fingerprint=session_data.device_public_key_fingerprint,
     )
 
     # Mark capture as completed in background

app/api/routes/device.py

@@ -86,6 +86,7 @@ async def create_device(
         session,
         publisher_id,
         request.external_id,
+        request.public_key,
         attestation_method=attestation.method,
         attested_at=attestation.attested_at,
         attested_app_id=attestation.app_id,

app/db/models.py

@@ -87,6 +87,9 @@ class Device(Base):
     attested_app_id: Mapped[str | None] = mapped_column(
         String(255), nullable=True, default=None
     )
+    # Content-signing public key (for cross-layer binding)
+    public_key: Mapped[str] = mapped_column(String(120))
+    device_public_key_fingerprint: Mapped[str] = mapped_column(String(64))
 
 
 class Capture(Base):

app/repositories/device.py

@@ -36,6 +36,8 @@ async def create(
         publisher_id: uuid.UUID,
         external_id: str,
         token_hash: str,
+        public_key: str,
+        device_public_key_fingerprint: str,
         attestation_method: str | None = None,
         attested_at: datetime | None = None,
         attested_app_id: str | None = None,
@@ -47,6 +49,8 @@ async def create(
             publisher_id: Publisher UUID.
             external_id: External device identifier.
             token_hash: Hashed device token.
+            public_key: Base64-encoded content-signing public key.
+            device_public_key_fingerprint: SHA-256 hex of the public key.
             attestation_method: Optional attestation method (e.g., "app_check").
             attested_at: Optional timestamp when attestation was verified.
             attested_app_id: Optional app ID from attestation (e.g., bundle ID).
@@ -62,6 +66,8 @@ async def create(
             attestation_method=attestation_method,
             attested_at=attested_at,
             attested_app_id=attested_app_id,
+            public_key=public_key,
+            device_public_key_fingerprint=device_public_key_fingerprint,
         )
         session.add(device)
         try:

app/schemas/device.py

@@ -1,4 +1,6 @@
-from pydantic import BaseModel, Field
+import base64
+
+from pydantic import BaseModel, Field, field_validator
 
 from app.schemas.base import APIDatetime, APIResponse
 
@@ -12,6 +14,26 @@ class DeviceCreateRequest(BaseModel):
         description="Unique identifier for the device (e.g., hardware ID, app installation ID)",
         json_schema_extra={"example": "device-abc-123"},
     )
+    public_key: str = Field(
+        min_length=1,
+        max_length=120,
+        description="Base64-encoded uncompressed EC public key (65 bytes: 0x04 + X + Y) from the device's content-signing key pair",
+    )
+
+    @field_validator("public_key")
+    @classmethod
+    def validate_public_key(cls, v: str) -> str:
+        try:
+            raw = base64.b64decode(v)
+        except Exception as e:
+            raise ValueError("Invalid Base64 encoding") from e
+        if len(raw) != 65:
+            raise ValueError(
+                f"Expected 65 bytes (uncompressed EC point), got {len(raw)}"
+            )
+        if raw[0] != 0x04:
+            raise ValueError("Expected uncompressed point format (0x04 prefix)")
+        return v
 
 
 class DeviceCreateResponse(APIResponse):

app/services/device.py

@@ -1,3 +1,4 @@
+import base64
 import hashlib
 import secrets
 import uuid
@@ -23,11 +24,18 @@ def _hash_token(self, token: str) -> str:
         """Hash a token for storage."""
         return hashlib.sha256(token.encode()).hexdigest()
 
+    @staticmethod
+    def _compute_fingerprint(public_key: str) -> str:
+        """Compute SHA-256 fingerprint of a Base64-encoded public key."""
+        raw_bytes = base64.b64decode(public_key)
+        return hashlib.sha256(raw_bytes).hexdigest()
+
     async def create(
         self,
         session: AsyncSession,
         publisher_id: uuid.UUID,
         external_id: str,
+        public_key: str,
         attestation_method: str | None = None,
         attested_at: datetime | None = None,
         attested_app_id: str | None = None,
@@ -39,6 +47,8 @@ async def create(
             session: Database session.
             publisher_id: Publisher UUID.
             external_id: External device identifier.
+            public_key: Base64-encoded uncompressed EC public key (65 bytes)
+                for cross-layer binding.
             attestation_method: Optional attestation method (e.g., "app_check").
             attested_at: Optional timestamp when attestation was verified.
             attested_app_id: Optional app ID from attestation (e.g., bundle ID).
@@ -49,6 +59,8 @@ async def create(
         token = self._generate_token()
         token_hash = self._hash_token(token)
 
+        fingerprint = self._compute_fingerprint(public_key)
+
         device = await self._repository.create(
             session,
             publisher_id,
@@ -57,6 +69,8 @@ async def create(
             attestation_method=attestation_method,
             attested_at=attested_at,
             attested_app_id=attested_app_id,
+            public_key=public_key,
+            device_public_key_fingerprint=fingerprint,
         )
 
         return device, token

app/services/session.py

@@ -20,6 +20,9 @@ class SessionData:
     device_id: str
     attestation_method: str | None  # None means no attestation (sandbox)
     app_id: str | None  # App ID from attestation (e.g., bundle ID)
+    device_public_key_fingerprint: (
+        str | None
+    )  # SHA-256 of device's content-signing public key
 
 
 @dataclass
@@ -53,6 +56,7 @@ async def create(
         device_id: uuid.UUID,
         attestation_method: str | None,
         app_id: str | None = None,
+        device_public_key_fingerprint: str | None = None,
     ) -> CreateSessionResult:
         """
         Create a new capture session for a device.
@@ -66,6 +70,8 @@ async def create(
             attestation_method: The device's attestation method (e.g., "app_check")
                 or None if not attested.
             app_id: The app ID from attestation (e.g., bundle ID).
+            device_public_key_fingerprint: SHA-256 hex of the device's content-signing
+                public key, for cross-layer binding in the JWT.
         """
         # Create capture record in database
         capture = await capture_repository.create(db, publisher_id, device_id)
@@ -83,6 +89,7 @@ async def create(
                 "device_id": str(device_id),
                 "attestation_method": attestation_method,
                 "app_id": app_id,
+                "device_public_key_fingerprint": device_public_key_fingerprint,
             }
         )
         await self._storage.set(
@@ -113,6 +120,7 @@ async def get_session_data(self, nonce: str) -> SessionData | None:
             device_id=data["device_id"],
             attestation_method=data.get("attestation_method"),
             app_id=data.get("app_id"),
+            device_public_key_fingerprint=data.get("device_public_key_fingerprint"),
         )
 
     async def consume(self, nonce: str) -> SessionData | None:

app/services/trust.py

@@ -32,6 +32,7 @@ def generate_token(
         device_id: str,
         method: AttestationMethod,
         app_id: str | None = None,
+        device_public_key_fingerprint: str | None = None,
     ) -> str:
         """
         Generate a signed JWT trust token.
@@ -42,6 +43,8 @@ def generate_token(
             device_id: The device ID.
             method: The attestation method used (sandbox, app_check, app_attest).
             app_id: The app ID from attestation (e.g., bundle ID), if available.
+            device_public_key_fingerprint: SHA-256 hex of the device's content-signing
+                public key, for cross-layer binding with the media integrity proof.
 
         Returns the signed JWT string with kid in header for JWKS lookup.
         """
@@ -63,6 +66,9 @@ def generate_token(
             "attestation": attestation,
         }
 
+        if device_public_key_fingerprint is not None:
+            payload["device_public_key_fingerprint"] = device_public_key_fingerprint
+
         return jwt.encode(
             payload,
             self._private_key,

tests/integration/test_capture_flow.py

@@ -26,7 +26,10 @@ def registered_device(
         """Create a device and return its credentials."""
         response = integration_client.post(
             "/devices",
-            json={"external_id": "capture-test-device"},
+            json={
+                "external_id": "capture-test-device",
+                "public_key": "BAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8gISIjJCUmJygpKissLS4vMDEyMzQ1Njc4OTo7PD0+P0A=",
+            },
             headers={"X-Publisher-ID": publisher_id},
         )
         assert response.status_code == status.HTTP_201_CREATED
@@ -193,12 +196,18 @@ def test_multiple_devices_isolated(
         # Create two devices
         device1 = integration_client.post(
             "/devices",
-            json={"external_id": "multi-device-001"},
+            json={
+                "external_id": "multi-device-001",
+                "public_key": "BAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8gISIjJCUmJygpKissLS4vMDEyMzQ1Njc4OTo7PD0+P0A=",
+            },
             headers={"X-Publisher-ID": publisher_id},
         ).json()
         device2 = integration_client.post(
             "/devices",
-            json={"external_id": "multi-device-002"},
+            json={
+                "external_id": "multi-device-002",
+                "public_key": "BAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8gISIjJCUmJygpKissLS4vMDEyMzQ1Njc4OTo7PD0+P0A=",
+            },
             headers={"X-Publisher-ID": publisher_id},
         ).json()
 

tests/integration/test_device_flow.py

@@ -26,7 +26,10 @@ def test_create_device_success(
         """Successfully create a new device."""
         response = integration_client.post(
             "/devices",
-            json={"external_id": "test-device-001"},
+            json={
+                "external_id": "test-device-001",
+                "public_key": "BAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8gISIjJCUmJygpKissLS4vMDEyMzQ1Njc4OTo7PD0+P0A=",
+            },
             headers={"X-Publisher-ID": publisher_id},
         )
 
@@ -46,15 +49,21 @@ def test_create_device_duplicate_fails(
         # Create first device
         response1 = integration_client.post(
             "/devices",
-            json={"external_id": "duplicate-device"},
+            json={
+                "external_id": "duplicate-device",
+                "public_key": "BAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8gISIjJCUmJygpKissLS4vMDEyMzQ1Njc4OTo7PD0+P0A=",
+            },
             headers={"X-Publisher-ID": publisher_id},
         )
         assert response1.status_code == status.HTTP_201_CREATED
 
         # Try to create again with same external_id
         response2 = integration_client.post(
             "/devices",
-            json={"external_id": "duplicate-device"},
+            json={
+                "external_id": "duplicate-device",
+                "public_key": "BAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8gISIjJCUmJygpKissLS4vMDEyMzQ1Njc4OTo7PD0+P0A=",
+            },
             headers={"X-Publisher-ID": publisher_id},
         )
         assert response2.status_code == status.HTTP_409_CONFLICT
@@ -67,7 +76,10 @@ def test_device_token_is_valid(
         # Create a device
         create_response = integration_client.post(
             "/devices",
-            json={"external_id": "session-test-device"},
+            json={
+                "external_id": "session-test-device",
+                "public_key": "BAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8gISIjJCUmJygpKissLS4vMDEyMzQ1Njc4OTo7PD0+P0A=",
+            },
             headers={"X-Publisher-ID": publisher_id},
         )
         assert create_response.status_code == status.HTTP_201_CREATED
@@ -107,15 +119,21 @@ def test_same_external_id_different_publishers(
         # Create device for first publisher
         device1_response = integration_client.post(
             "/devices",
-            json={"external_id": shared_external_id},
+            json={
+                "external_id": shared_external_id,
+                "public_key": "BAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8gISIjJCUmJygpKissLS4vMDEyMzQ1Njc4OTo7PD0+P0A=",
+            },
             headers={"X-Publisher-ID": publisher_id_1},
         )
         assert device1_response.status_code == status.HTTP_201_CREATED
 
         # Create device with SAME external_id for second publisher - should succeed
         device2_response = integration_client.post(
             "/devices",
-            json={"external_id": shared_external_id},
+            json={
+                "external_id": shared_external_id,
+                "public_key": "BAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8gISIjJCUmJygpKissLS4vMDEyMzQ1Njc4OTo7PD0+P0A=",
+            },
             headers={"X-Publisher-ID": publisher_id_2},
         )
         assert device2_response.status_code == status.HTTP_201_CREATED
@@ -273,7 +291,10 @@ def test_device_created_at_format(self, integration_client: TestClient) -> None:
         # Create device
         device_response = integration_client.post(
             "/devices",
-            json={"external_id": "datetime-test-device"},
+            json={
+                "external_id": "datetime-test-device",
+                "public_key": "BAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8gISIjJCUmJygpKissLS4vMDEyMzQ1Njc4OTo7PD0+P0A=",
+            },
             headers={"X-Publisher-ID": publisher_id},
         )
         assert device_response.status_code == status.HTTP_201_CREATED

tests/integration/test_jwks.py

@@ -36,7 +36,10 @@ def test_trust_token_has_kid_matching_jwks(
 
         dev_response = integration_client.post(
             "/devices",
-            json={"external_id": "jwks-test-device"},
+            json={
+                "external_id": "jwks-test-device",
+                "public_key": "BAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8gISIjJCUmJygpKissLS4vMDEyMzQ1Njc4OTo7PD0+P0A=",
+            },
             headers={"X-Publisher-ID": publisher_id},
         )
         device_token = dev_response.json()["device_token"]
@@ -80,7 +83,10 @@ def test_trust_token_can_be_verified_with_jwks(
 
         dev_response = integration_client.post(
             "/devices",
-            json={"external_id": "jwks-verify-device"},
+            json={
+                "external_id": "jwks-verify-device",
+                "public_key": "BAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8gISIjJCUmJygpKissLS4vMDEyMzQ1Njc4OTo7PD0+P0A=",
+            },
             headers={"X-Publisher-ID": publisher_id},
         )
         device_token = dev_response.json()["device_token"]