feat(verifier): add local Ed25519 attestation verifier

Author: josedab

streamline_sdk/attestation.py

@@ -0,0 +1,214 @@
+"""Attestation client (M4 wiring).
+
+Wraps ``POST /api/v1/attest`` and ``POST /api/v1/attest/verify`` for
+producing and verifying Ed25519 envelope signatures over Kafka records.
+
+Example:
+    attestor = Attestor("http://localhost:9094", key_id="broker-0")
+    sig = await attestor.sign(
+        topic="orders", partition=0, offset=42,
+        value=b'{"id":1}', schema_id=7,
+    )
+    ok = await attestor.verify(
+        topic="orders", partition=0, offset=42,
+        value=b'{"id":1}', schema_id=7,
+        timestamp_ms=sig.timestamp_ms, signature_b64=sig.signature_b64,
+    )
+"""
+
+from __future__ import annotations
+
+import asyncio
+import base64
+import json
+import time
+from dataclasses import dataclass
+from typing import Any, Optional, Union
+
+try:
+    import aiohttp
+
+    _HAS_AIOHTTP = True
+except ImportError:  # pragma: no cover
+    _HAS_AIOHTTP = False
+
+from .exceptions import StreamlineError
+
+ATTEST_HEADER = "streamline-attest"
+
+
+class AttestationError(StreamlineError):
+    """Raised on attestation API failures."""
+
+
+@dataclass
+class SignedAttestation:
+    """Result of a successful sign request."""
+
+    key_id: str
+    algorithm: str
+    timestamp_ms: int
+    payload_sha256: str
+    signature_b64: str
+    header_name: str
+    header_value: str
+
+    @classmethod
+    def from_json(cls, data: dict[str, Any]) -> "SignedAttestation":
+        return cls(
+            key_id=str(data["key_id"]),
+            algorithm=str(data["algorithm"]),
+            timestamp_ms=int(data["timestamp_ms"]),
+            payload_sha256=str(data["payload_sha256"]),
+            signature_b64=str(data["signature_b64"]),
+            header_name=str(data["header_name"]),
+            header_value=str(data["header_value"]),
+        )
+
+
+def _encode_value(
+    value: Union[bytes, bytearray, str],
+) -> tuple[str, str]:
+    """Return (json_field_name, json_field_value) for the value."""
+    if isinstance(value, (bytes, bytearray)):
+        return "value_b64", base64.b64encode(bytes(value)).decode("ascii")
+    return "value", str(value)
+
+
+class Attestor:
+    """Async client for the broker's attestation sign/verify routes.
+
+    Args:
+        http_url: HTTP base URL.
+        key_id: Default signing key id used by :meth:`sign`.
+        algorithm: Default signature algorithm (server currently supports
+            ``ed25519``; ``ecdsa-p256`` may also be accepted by ``verify``).
+        timeout: Per-request timeout in seconds.
+    """
+
+    def __init__(
+        self,
+        http_url: str = "http://localhost:9094",
+        key_id: str = "broker-0",
+        algorithm: str = "ed25519",
+        timeout: float = 10.0,
+    ) -> None:
+        self.http_url = http_url.rstrip("/")
+        self.key_id = key_id
+        self.algorithm = algorithm
+        self.timeout = timeout
+
+    async def sign(
+        self,
+        *,
+        topic: str,
+        partition: int,
+        offset: int,
+        value: Union[bytes, bytearray, str],
+        schema_id: int = 0,
+        timestamp_ms: Optional[int] = None,
+        key_id: Optional[str] = None,
+    ) -> SignedAttestation:
+        """Sign an attestation envelope for a record."""
+        ts = timestamp_ms if timestamp_ms is not None else int(time.time() * 1000)
+        body: dict[str, Any] = {
+            "topic": topic,
+            "partition": partition,
+            "offset": offset,
+            "schema_id": schema_id,
+            "timestamp_ms": ts,
+            "key_id": key_id or self.key_id,
+        }
+        field, encoded = _encode_value(value)
+        body[field] = encoded
+
+        status, payload = await self._post("/api/v1/attest", body)
+        if status != 200:
+            raise AttestationError(
+                f"sign -> HTTP {status}: {json.dumps(payload)[:512]}"
+            )
+        return SignedAttestation.from_json(payload or {})
+
+    async def verify(
+        self,
+        *,
+        topic: str,
+        partition: int,
+        offset: int,
+        value: Union[bytes, bytearray, str],
+        timestamp_ms: int,
+        signature_b64: str,
+        schema_id: int = 0,
+        key_id: Optional[str] = None,
+        algorithm: Optional[str] = None,
+    ) -> bool:
+        """Verify a previously-issued attestation. Returns ``True`` on success."""
+        body: dict[str, Any] = {
+            "topic": topic,
+            "partition": partition,
+            "offset": offset,
+            "schema_id": schema_id,
+            "timestamp_ms": timestamp_ms,
+            "key_id": key_id or self.key_id,
+            "signature_b64": signature_b64,
+            "algorithm": algorithm or self.algorithm,
+        }
+        field, encoded = _encode_value(value)
+        body[field] = encoded
+
+        status, payload = await self._post("/api/v1/attest/verify", body)
+        if status != 200:
+            raise AttestationError(
+                f"verify -> HTTP {status}: {json.dumps(payload)[:512]}"
+            )
+        return bool((payload or {}).get("valid", False))
+
+    async def _post(
+        self, path: str, body: dict[str, Any]
+    ) -> tuple[int, Any]:
+        url = f"{self.http_url}{path}"
+        if _HAS_AIOHTTP:
+            timeout = aiohttp.ClientTimeout(total=self.timeout)
+            async with aiohttp.ClientSession(timeout=timeout) as session:
+                async with session.post(url, json=body) as resp:
+                    text = await resp.text()
+                    payload: Any = None
+                    if text:
+                        try:
+                            payload = json.loads(text)
+                        except json.JSONDecodeError:
+                            payload = {"raw": text}
+                    return resp.status, payload
+        else:
+            import urllib.request
+            import urllib.error
+
+            data = json.dumps(body).encode("utf-8")
+            req = urllib.request.Request(
+                url,
+                data=data,
+                headers={"Content-Type": "application/json"},
+                method="POST",
+            )
+
+            def _sync() -> tuple[int, Any]:
+                try:
+                    with urllib.request.urlopen(req, timeout=self.timeout) as resp:
+                        text = resp.read().decode("utf-8")
+                        return resp.status, json.loads(text) if text else None
+                except urllib.error.HTTPError as e:
+                    text = e.read().decode("utf-8", errors="replace")
+                    try:
+                        return e.code, json.loads(text)
+                    except json.JSONDecodeError:
+                        return e.code, {"raw": text}
+
+            return await asyncio.to_thread(_sync)
+
+
+__all__ = [
+    "ATTEST_HEADER",
+    "Attestor",
+    "AttestationError",
+    "SignedAttestation",
+]

streamline_sdk/verifier.py

@@ -0,0 +1,119 @@
+"""Local Ed25519 attestation verifier for Streamline consumer records.
+
+Verifies ``streamline-attest`` headers locally using an Ed25519 public key
+without any network calls.
+"""
+
+from __future__ import annotations
+
+import base64
+import json
+from dataclasses import dataclass
+from typing import Any, Dict, Optional, Union
+
+from cryptography.exceptions import InvalidSignature
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
+
+ATTEST_HEADER = "streamline-attest"
+
+
+@dataclass
+class VerificationResult:
+    """Result of an attestation verification."""
+
+    verified: bool
+    producer_id: str = ""
+    schema_id: Optional[int] = None
+    contract_id: Optional[str] = None
+    timestamp_ms: int = 0
+
+
+class StreamlineVerifier:
+    """Verifies attestation headers on consumed records using a local Ed25519 public key.
+
+    Args:
+        public_key: An Ed25519 public key used for signature verification.
+
+    Example::
+
+        from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
+        from streamline_sdk.verifier import StreamlineVerifier
+
+        pub_key = Ed25519PublicKey.from_public_bytes(key_bytes)
+        verifier = StreamlineVerifier(pub_key)
+        result = verifier.verify(record)
+        if result.verified:
+            print(f"Verified from {result.producer_id}")
+    """
+
+    def __init__(self, public_key: Ed25519PublicKey) -> None:
+        self._public_key = public_key
+
+    def verify(self, record: Any) -> VerificationResult:
+        """Verify the attestation on a consumer record.
+
+        Extracts the ``streamline-attest`` header, parses the base64-encoded
+        JSON attestation, reconstructs the canonical bytes, and verifies the
+        Ed25519 signature.
+
+        Args:
+            record: A ConsumerRecord with a ``headers`` dict containing
+                the ``streamline-attest`` header.
+
+        Returns:
+            VerificationResult with ``verified=True`` if the signature is valid.
+        """
+        headers: Dict[str, Union[bytes, str]] = getattr(record, "headers", {})
+        raw = headers.get(ATTEST_HEADER)
+        if raw is None:
+            return VerificationResult(verified=False)
+
+        try:
+            raw_bytes = raw if isinstance(raw, (bytes, bytearray)) else raw.encode("utf-8")
+            attestation = json.loads(base64.b64decode(raw_bytes))
+        except (json.JSONDecodeError, Exception):
+            return VerificationResult(verified=False)
+
+        try:
+            payload_sha256 = str(attestation["payload_sha256"])
+            topic = str(attestation["topic"])
+            partition = int(attestation["partition"])
+            offset = int(attestation["offset"])
+            schema_id = int(attestation["schema_id"])
+            timestamp_ms = int(attestation["timestamp_ms"])
+            key_id = str(attestation["key_id"])
+            signature_b64 = str(attestation["signature"])
+        except (KeyError, ValueError, TypeError):
+            return VerificationResult(verified=False)
+
+        canonical = (
+            f"{topic}|{partition}|{offset}|{payload_sha256}"
+            f"|{schema_id}|{timestamp_ms}|{key_id}"
+        )
+        canonical_bytes = canonical.encode("utf-8")
+
+        try:
+            signature = base64.b64decode(signature_b64)
+        except Exception:
+            return VerificationResult(verified=False)
+
+        try:
+            self._public_key.verify(signature, canonical_bytes)
+            verified = True
+        except InvalidSignature:
+            verified = False
+
+        return VerificationResult(
+            verified=verified,
+            producer_id=key_id,
+            schema_id=schema_id if schema_id != 0 else None,
+            contract_id=attestation.get("contract_id"),
+            timestamp_ms=timestamp_ms,
+        )
+
+
+__all__ = [
+    "ATTEST_HEADER",
+    "StreamlineVerifier",
+    "VerificationResult",
+]