Golden Age Hub: Phase 1 Technical Specification & Architecture v2.0¶

from datetime import datetime, timedelta
from functools import lru_cache
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey

class KeyManager:
    @lru_cache(maxsize=10000)
    def get_active_key(self, device_id: str) -> Ed25519PublicKey:
        """Cached lookup of active public key for device"""
        now = datetime.utcnow()
        key_record = db.query(DeviceKey).filter(
            DeviceKey.device_id == device_id,
            DeviceKey.valid_from <= now,
            DeviceKey.valid_until >= now
        ).first()

        if not key_record:
            raise InvalidDeviceKeyError(f"No valid key for {device_id}")

        return Ed25519PublicKey.from_public_bytes(key_record.public_key)

    async def rotate_key(self, device_id: str, new_public_key: bytes,
                         old_signature: bytes, payload: bytes) -> bool:
        """Verify old signature, register new key with overlapping validity"""
        # Verify device owns the current key
        current_key = self.get_active_key(device_id)
        try:
            current_key.verify(old_signature, payload)
        except Exception:
            raise UnauthorizedKeyRotation()

        # Store new key with overlapping validity
        new_key = DeviceKey(
            device_id=device_id,
            public_key=new_public_key,
            key_version=current_key_record.key_version + 1,
            valid_from=datetime.utcnow() + timedelta(days=7),  # 7 day transition
            valid_until=datetime.utcnow() + timedelta(days=97)  # 90 + 7
        )
        db.add(new_key)
        db.commit()

        # Invalidate cache
        self.get_active_key.cache_clear()
        return True

import jcs  # RFC 8785 implementation

def canonicalize_event(event_payload: dict) -> bytes:
    """
    Convert event to canonical JSON form for signature verification.
    The signature field is excluded from canonicalization.
    """
    # Create a copy without the signature field
    canonical_payload = {k: v for k, v in event_payload.items() if k != 'signature'}

    # JCS canonicalization (RFC 8785)
    canonical_bytes = jcs.canonicalize(canonical_payload)

    return canonical_bytes

import asyncio
from concurrent.futures import ThreadPoolExecutor
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
from cryptography.exceptions import InvalidSignature

# Dedicated executor for CPU-intensive crypto operations
crypto_executor = ThreadPoolExecutor(max_workers=4, thread_name_prefix="crypto-worker")

async def verify_signature_async(event: EventPayload, key_manager: KeyManager) -> bool:
    """
    Asynchronously verify Ed25519 signature without blocking event loop.
    Returns True if valid, raises exception otherwise.
    """
    loop = asyncio.get_event_loop()

    # Offload to thread pool
    result = await loop.run_in_executor(
        crypto_executor,
        _verify_signature_blocking,
        event,
        key_manager
    )

    return result

def _verify_signature_blocking(event: EventPayload, key_manager: KeyManager) -> bool:
    """Blocking signature verification (runs in thread pool)"""
    try:
        # Get device's public key
        public_key = key_manager.get_active_key(event.device_id)

        # Canonicalize payload (excluding signature)
        canonical_bytes = canonicalize_event(event.dict())

        # Decode signature from hex string
        signature_bytes = bytes.fromhex(event.signature)

        # Verify Ed25519 signature
        public_key.verify(signature_bytes, canonical_bytes)

        return True
    except InvalidSignature:
        raise SignatureVerificationError("Invalid signature")
    except Exception as e:
        raise SignatureVerificationError(f"Verification failed: {str(e)}")

class ReplayProtection:
    def __init__(self):
        # In-memory cache of last seen sequence per device
        self.last_sequences = {}  # {device_id: sequence_id}

    async def validate_sequence(self, device_id: str, sequence_id: int) -> bool:
        """
        Validate that sequence_id is strictly greater than last seen.
        Prevents replay attacks and detects out-of-order delivery.
        """
        # Get last known sequence from cache or database
        last_seq = self.last_sequences.get(device_id)

        if last_seq is None:
            # First event from this device, query database
            last_event = await db.query(Event).filter(
                Event.device_id == device_id
            ).order_by(Event.sequence_id.desc()).first()

            last_seq = last_event.sequence_id if last_event else -1
            self.last_sequences[device_id] = last_seq

        # Sequence must be strictly increasing
        if sequence_id <= last_seq:
            raise ReplayAttackDetected(
                f"Device {device_id} sent sequence {sequence_id}, "
                f"but last seen was {last_seq}"
            )

        # Update cache
        self.last_sequences[device_id] = sequence_id
        return True

    async def handle_sequence_gap(self, device_id: str, expected: int, received: int):
        """
        Log sequence gaps (potential offline period or packet loss).
        Gaps are allowed but logged for monitoring.
        """
        gap_size = received - expected
        await db.execute(
            """
            INSERT INTO sequence_gaps (device_id, expected_seq, received_seq, gap_size, detected_at)
            VALUES (:device, :expected, :received, :gap, NOW())
            """,
            device=device_id, expected=expected, received=received, gap=gap_size
        )

        # Alert if gap is suspiciously large (possible attack or device replacement)
        if gap_size > 10000:
            await alert_security_team(device_id, gap_size)

def validate_timestamp(event_timestamp: datetime) -> bool:
    server_time = datetime.utcnow()
    delta = abs((event_timestamp - server_time).total_seconds())

    if delta > 300:  # 5 minutes
        logger.warning(
            f"Clock skew detected: {delta}s",
            extra={"device_id": event.device_id, "event_id": event.event_id}
        )
        # Still accept event (clocks drift, NTP may be unavailable)

    return True

from pydantic import BaseModel, Field

class EventPayloadV1_0(BaseModel):
    event_id: UUID
    patient_id: UUID
    device_id: str = Field(pattern=r"^GA-[A-Z0-9]{6}$")
    # ... v1.0 fields ...

class EventPayloadV1_1(EventPayloadV1_0):  # Inherits v1.0
    battery_level: Optional[float] = Field(None, ge=0, le=100)

def validate_event(payload: dict) -> BaseModel:
    schema_version = payload.get('_schema_version', '1.0.0')

    if schema_version.startswith('1.0'):
        return EventPayloadV1_0(**payload)
    elif schema_version.startswith('1.1'):
        return EventPayloadV1_1(**payload)
    else:
        raise UnsupportedSchemaVersion(
            f"Schema version {schema_version} not supported. "
            f"Supported versions: 1.0.x, 1.1.x"
        )

from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse
import time

app = FastAPI()

@app.post("/v1/events/ingest")
async def ingest_event(request: Request, event: dict):
    """
    Main ingestion endpoint with comprehensive security checks.
    """
    try:
        # Step 1: Rate Limiting (per-device token bucket)
        device_id = event.get('device_id')
        if not await rate_limiter.check_and_consume(device_id):
            raise HTTPException(429, "Rate limit exceeded")

        # Step 2: Schema Validation
        validated_event = validate_event(event)

        # Step 3: Signature Verification (async, offloaded to thread pool)
        if not await verify_signature_async(validated_event, key_manager):
            await alert_security_team("invalid_signature", device_id)
            raise HTTPException(401, "Invalid signature")

        # Step 4: Replay Protection
        if not await replay_protection.validate_sequence(
            validated_event.device_id,
            validated_event.sequence_id
        ):
            await alert_security_team("replay_detected", device_id)
            raise HTTPException(409, "Replay attack detected")

        # Step 5: Queue for async processing
        await redis_queue.enqueue(validated_event.json())

        return JSONResponse(
            status_code=202,
            content={
                "status": "accepted",
                "event_id": str(validated_event.event_id),
                "message": "Event queued for processing"
            }
        )

    except HTTPException:
        raise  # Re-raise HTTP errors as-is
    except Exception as e:
        logger.error(f"Unexpected ingestion error: {e}", exc_info=True)
        raise HTTPException(500, "Internal server error")

class TokenBucketRateLimiter:
    def __init__(self, rate: float = 10.0, burst_size: int = 100):
        """
        Args:
            rate: Average tokens per second (e.g., 10 = 36,000 events/hour)
            burst_size: Maximum burst capacity (e.g., 100 = sync after ~10s offline)
        """
        self.rate = rate
        self.burst_size = burst_size
        self.redis = redis.StrictRedis()

    async def check_and_consume(self, device_id: str) -> bool:
        """
        Check if device has tokens available, consume 1 if so.
        Uses Redis for distributed rate limiting across API servers.
        """
        key = f"ratelimit:{device_id}"
        now = time.time()

        # Lua script for atomic token bucket operation
        script = """
        local key = KEYS[1]
        local rate = tonumber(ARGV[1])
        local burst = tonumber(ARGV[2])
        local now = tonumber(ARGV[3])

        local bucket = redis.call('HMGET', key, 'tokens', 'last_update')
        local tokens = tonumber(bucket[1]) or burst
        local last_update = tonumber(bucket[2]) or now

        -- Refill tokens based on elapsed time
        local elapsed = now - last_update
        tokens = math.min(burst, tokens + (elapsed * rate))

        -- Try to consume 1 token
        if tokens >= 1 then
            tokens = tokens - 1
            redis.call('HMSET', key, 'tokens', tokens, 'last_update', now)
            redis.call('EXPIRE', key, 3600)  -- Expire after 1 hour of inactivity
            return 1
        else
            return 0
        end
        """

        result = await self.redis.eval(
            script,
            1,
            key,
            self.rate,
            self.burst_size,
            now
        )

        return result == 1

import asyncio
from typing import Optional

class PipelineWorker:
    def __init__(self, worker_id: int, redis_url: str, concurrency: int = 10):
        self.worker_id = worker_id
        self.redis = aioredis.from_url(redis_url)
        self.concurrency = concurrency
        self.running = False

    async def start(self):
        """Start worker with concurrent event processing"""
        self.running = True
        logger.info(f"Worker {self.worker_id} started with concurrency={self.concurrency}")

        # Create pool of concurrent processors
        tasks = [
            asyncio.create_task(self._process_loop())
            for _ in range(self.concurrency)
        ]

        await asyncio.gather(*tasks)

    async def _process_loop(self):
        """Main processing loop for a single concurrent processor"""
        while self.running:
            try:
                # Blocking pop with timeout (BRPOP for FIFO queue)
                result = await self.redis.brpop('event_queue', timeout=5)

                if result:
                    queue_name, event_json = result
                    await self.process_event(event_json)

            except asyncio.CancelledError:
                break
            except Exception as e:
                logger.error(f"Worker {self.worker_id} error: {e}", exc_info=True)
                await asyncio.sleep(1)  # Back off on errors

    async def process_event(self, event_json: str):
        """
        Process a single event: normalize to PostgreSQL + archive to S3.
        Implements retry logic with exponential backoff.
        """
        retry_count = 0
        max_retries = 3

        while retry_count <= max_retries:
            try:
                # Parse event
                event_data = json.loads(event_json)
                event = EventPayload(**event_data)

                # Dual-write: PostgreSQL (normalized) + S3 (raw archive)
                await asyncio.gather(
                    self.store_normalized(event),
                    self.archive_to_s3(event)
                )

                # Success metrics
                events_processed.inc()
                return

            except Exception as e:
                retry_count += 1
                if retry_count > max_retries:
                    # Move to Dead Letter Queue
                    await self.send_to_dlq(event_json, str(e))
                    events_failed.inc()
                    logger.error(
                        f"Event {event_data.get('event_id')} failed after {max_retries} retries",
                        extra={"error": str(e), "event": event_json}
                    )
                    return

                # Exponential backoff: 1s, 2s, 4s
                backoff = 2 ** (retry_count - 1)
                logger.warning(
                    f"Event processing failed, retry {retry_count}/{max_retries} in {backoff}s",
                    extra={"error": str(e)}
                )
                await asyncio.sleep(backoff)

    async def store_normalized(self, event: EventPayload):
        """Store event in PostgreSQL/TimescaleDB normalized format"""
        async with db.transaction():
            await db.execute(
                """
                INSERT INTO events (
                    event_id, patient_id, device_id, sequence_id,
                    event_timestamp, event_type, confidence,
                    location_lat, location_lng, location_accuracy_m,
                    sensor_data, raw_payload
                ) VALUES (
                    :event_id, :patient_id, :device_id, :sequence_id,
                    :timestamp, :event_type, :confidence,
                    :lat, :lng, :accuracy,
                    :sensor_data, :raw_payload
                )
                """,
                event_id=event.event_id,
                patient_id=event.patient_id,
                device_id=event.device_id,
                sequence_id=event.sequence_id,
                timestamp=event.timestamp,
                event_type=event.event_type,
                confidence=event.confidence,
                lat=event.location.lat if event.location else None,
                lng=event.location.lng if event.location else None,
                accuracy=event.location.accuracy_m if event.location else None,
                sensor_data=json.dumps(event.sensor_data) if event.sensor_data else {},
                raw_payload=event.json()
            )

    async def archive_to_s3(self, event: EventPayload):
        """Archive original signed payload to S3 for compliance"""
        # Partition by date for efficient querying
        date_partition = event.timestamp.strftime("%Y/%m/%d")
        key = f"events/{date_partition}/{event.event_id}.json"

        await s3_client.put_object(
            Bucket=ARCHIVE_BUCKET,
            Key=key,
            Body=event.json(),
            ContentType='application/json',
            Metadata={
                'patient_id': str(event.patient_id),
                'device_id': event.device_id,
                'event_type': event.event_type,
                'ingested_at': datetime.utcnow().isoformat()
            },
            ServerSideEncryption='AES256',
            ObjectLockMode='COMPLIANCE',  # WORM
            ObjectLockRetainUntilDate=datetime.utcnow() + timedelta(days=2555)  # 7 years
        )

    async def send_to_dlq(self, event_json: str, error: str):
        """Send failed event to Dead Letter Queue for manual intervention"""
        dlq_entry = {
            'event': json.loads(event_json),
            'error': error,
            'failed_at': datetime.utcnow().isoformat(),
            'worker_id': self.worker_id
        }

        await self.redis.lpush('dead_letter_queue', json.dumps(dlq_entry))

        # Alert operations team for DLQ entries
        await alert_operations(
            severity="warning",
            title="Event Processing Failed",
            description=f"Event moved to DLQ: {error}",
            event_id=dlq_entry['event'].get('event_id')
        )

    async def stop(self):
        """Graceful shutdown"""
        self.running = False
        logger.info(f"Worker {self.worker_id} stopping...")

@app.get("/v1/admin/dlq")
async def get_dlq_stats(current_user: User = Depends(require_admin)):
    """Get Dead Letter Queue statistics and failed events"""
    dlq_size = await redis.llen('dead_letter_queue')

    # Get last 100 failed events
    failed_events = await redis.lrange('dead_letter_queue', 0, 99)

    # Parse and summarize
    failures_by_type = {}
    for entry_json in failed_events:
        entry = json.loads(entry_json)
        error_type = entry['error'].split(':')[0]  # First part of error message
        failures_by_type[error_type] = failures_by_type.get(error_type, 0) + 1

    return {
        'dlq_size': dlq_size,
        'failures_by_type': failures_by_type,
        'sample_failures': [json.loads(e) for e in failed_events[:10]]
    }

@app.post("/v1/admin/dlq/{event_id}/retry")
async def retry_dlq_event(event_id: str, current_user: User = Depends(require_admin)):
    """Manually retry a failed event from DLQ"""
    # Find event in DLQ
    dlq_events = await redis.lrange('dead_letter_queue', 0, -1)

    for entry_json in dlq_events:
        entry = json.loads(entry_json)
        if entry['event']['event_id'] == event_id:
            # Re-queue for processing
            await redis.lpush('event_queue', json.dumps(entry['event']))

            # Remove from DLQ
            await redis.lrem('dead_letter_queue', 1, entry_json)

            return {"status": "requeued", "event_id": event_id}

    raise HTTPException(404, "Event not found in DLQ")

class BackpressureManager:
    def __init__(self, redis: aioredis.Redis, max_queue_depth: int = 10000):
        self.redis = redis
        self.max_queue_depth = max_queue_depth

    async def check_capacity(self) -> bool:
        """Check if system can accept more events"""
        queue_depth = await self.redis.llen('event_queue')

        if queue_depth >= self.max_queue_depth:
            # Queue is full, reject new events
            queue_full_rejections.inc()
            return False

        # Update metrics
        queue_depth_gauge.set(queue_depth)
        return True

# In ingestion endpoint:
@app.post("/v1/events/ingest")
async def ingest_event(event: dict):
    # ... (previous validation steps) ...

    # Check backpressure before queueing
    if not await backpressure.check_capacity():
        raise HTTPException(
            status_code=503,
            detail="System at capacity, please retry later",
            headers={"Retry-After": "60"}  # Suggest retry in 60 seconds
        )

    await redis_queue.enqueue(validated_event.json())
    return JSONResponse(status_code=202, content={"status": "accepted"})

# migrations/versions/001_initial_schema.py
"""Initial schema for events and device keys"""

from alembic import op
import sqlalchemy as sa

def upgrade():
    op.execute("CREATE EXTENSION IF NOT EXISTS timescaledb")

    op.create_table(
        'events',
        sa.Column('event_id', sa.UUID(), primary_key=True),
        # ... (full schema from above)
    )

    op.execute(
        "SELECT create_hypertable('events', 'event_timestamp', "
        "chunk_time_interval => INTERVAL '1 day')"
    )

def downgrade():
    op.drop_table('events')

# migrations/versions/002_add_battery_level.py
"""Add optional battery_level for schema v1.1"""

def upgrade():
    op.add_column('events',
        sa.Column('battery_level', sa.Float(), nullable=True,
                  comment='Added in schema v1.1')
    )

def downgrade():
    op.drop_column('events', 'battery_level')

from typing import Optional
from pydantic import BaseModel, Field, validator

class EventPayloadV1_0(BaseModel):
    """Schema v1.0 - Initial frozen contract"""
    event_id: UUID
    patient_id: UUID
    device_id: str
    sequence_id: int
    timestamp: datetime
    event_type: str
    confidence: Optional[float]
    location: Optional[Location]
    sensor_data: Optional[dict]
    signature: str

class EventPayloadV1_1(EventPayloadV1_0):
    """Schema v1.1 - Adds optional battery monitoring"""
    battery_level: Optional[float] = Field(
        None,
        ge=0,
        le=100,
        description="Device battery percentage. Added in v1.1."
    )

    @validator('battery_level')
    def validate_battery(cls, v):
        if v is not None and (v < 0 or v > 100):
            raise ValueError("battery_level must be between 0 and 100")
        return v

class EventPayloadV1_2(EventPayloadV1_1):
    """Schema v1.2 - Adds optional network quality metrics"""
    network_rssi: Optional[int] = Field(
        None,
        ge=-120,
        le=0,
        description="Network signal strength in dBm. Added in v1.2."
    )

# Schema registry for multi-version support
SCHEMA_REGISTRY = {
    '1.0': EventPayloadV1_0,
    '1.0.0': EventPayloadV1_0,
    '1.1': EventPayloadV1_1,
    '1.1.0': EventPayloadV1_1,
    '1.2': EventPayloadV1_2,
    '1.2.0': EventPayloadV1_2,
}

def validate_event(payload: dict) -> BaseModel:
    """Validate event against appropriate schema version"""
    schema_version = payload.get('_schema_version', '1.0.0')

    # Normalize version string (1.1 -> 1.1.0)
    if schema_version.count('.') == 1:
        schema_version += '.0'

    schema_class = SCHEMA_REGISTRY.get(schema_version)
    if not schema_class:
        raise UnsupportedSchemaVersion(
            f"Schema version {schema_version} not supported. "
            f"Supported versions: {', '.join(SCHEMA_REGISTRY.keys())}"
        )

    try:
        return schema_class(**payload)
    except ValidationError as e:
        raise SchemaValidationError(f"Invalid payload for schema {schema_version}: {e}")