from fastapi import APIRouter, Depends, Query, UploadFile, File, Form, HTTPException from fastapi.responses import FileResponse, PlainTextResponse from pydantic import BaseModel from typing import Optional import logging from sqlalchemy.ext.asyncio import AsyncSession from auth.models import TokenPayload from auth.dependencies import require_permission from firmware.models import FirmwareVersion, FirmwareListResponse, FirmwareMetadataResponse, UpdateType from firmware import service from database.postgres import get_pg_session from shared.audit import log_action logger = logging.getLogger(__name__) router = APIRouter(prefix="/api/firmware", tags=["firmware"]) ota_router = APIRouter(prefix="/api/ota", tags=["ota-telemetry"]) @router.post("/upload", response_model=FirmwareVersion, status_code=201) async def upload_firmware( hw_type: str = Form(...), channel: str = Form(...), version: str = Form(...), update_type: UpdateType = Form(UpdateType.mandatory), min_fw_version: Optional[str] = Form(None), changelog: Optional[str] = Form(None), release_note: Optional[str] = Form(None), bespoke_uid: Optional[str] = Form(None), file: UploadFile = File(...), _user: TokenPayload = Depends(require_permission("manufacturing", "add")), db: AsyncSession = Depends(get_pg_session), ): file_bytes = await file.read() fw = service.upload_firmware( hw_type=hw_type, channel=channel, version=version, file_bytes=file_bytes, update_type=update_type, min_fw_version=min_fw_version, changelog=changelog, release_note=release_note, bespoke_uid=bespoke_uid, ) await log_action(db, _user.sub, _user.name or _user.email, "CREATE", "firmware", fw.id, f"{hw_type} v{version} ({channel})") return fw @router.get("", response_model=FirmwareListResponse) def list_firmware( hw_type: Optional[str] = Query(None), channel: Optional[str] = Query(None), _user: TokenPayload = Depends(require_permission("manufacturing", "view")), ): items = service.list_firmware(hw_type=hw_type, channel=channel) return FirmwareListResponse(firmware=items, total=len(items)) @router.get("/{hw_type}/{channel}/latest", response_model=FirmwareMetadataResponse) def get_latest_firmware( hw_type: str, channel: str, hw_version: Optional[str] = Query(None, description="Hardware revision from NVS, e.g. '1.0'"), current_version: Optional[str] = Query(None, description="Currently running firmware semver, e.g. '1.2.3'"), ): """Returns metadata for the latest firmware for a given hw_type + channel. No auth required — devices call this endpoint to check for updates. """ return service.get_latest(hw_type, channel, hw_version=hw_version, current_version=current_version) @router.get("/{hw_type}/{channel}/latest/changelog", response_class=PlainTextResponse) def get_latest_changelog(hw_type: str, channel: str): """Returns the full changelog for the latest firmware. Plain text.""" return service.get_latest_changelog(hw_type, channel) @router.get("/{hw_type}/{channel}/{version}/info/changelog", response_class=PlainTextResponse) def get_version_changelog(hw_type: str, channel: str, version: str): """Returns the full changelog for a specific firmware version. Plain text.""" return service.get_version_changelog(hw_type, channel, version) @router.get("/{hw_type}/{channel}/{version}/info", response_model=FirmwareMetadataResponse) def get_firmware_info(hw_type: str, channel: str, version: str): """Returns metadata for a specific firmware version. No auth required — devices call this to resolve upgrade chains. """ return service.get_version_info(hw_type, channel, version) @router.get("/{hw_type}/{channel}/{version}/firmware.bin") def download_firmware(hw_type: str, channel: str, version: str): """Download the firmware binary. No auth required — devices call this directly.""" path = service.get_firmware_path(hw_type, channel, version) return FileResponse( path=str(path), media_type="application/octet-stream", filename="firmware.bin", ) @router.put("/{firmware_id}", response_model=FirmwareVersion) async def edit_firmware( firmware_id: str, channel: Optional[str] = Form(None), version: Optional[str] = Form(None), update_type: Optional[UpdateType] = Form(None), min_fw_version: Optional[str] = Form(None), changelog: Optional[str] = Form(None), release_note: Optional[str] = Form(None), bespoke_uid: Optional[str] = Form(None), file: Optional[UploadFile] = File(None), _user: TokenPayload = Depends(require_permission("manufacturing", "add")), db: AsyncSession = Depends(get_pg_session), ): file_bytes = await file.read() if file and file.filename else None fw = service.edit_firmware( doc_id=firmware_id, channel=channel, version=version, update_type=update_type, min_fw_version=min_fw_version, changelog=changelog, release_note=release_note, bespoke_uid=bespoke_uid, file_bytes=file_bytes, ) await log_action(db, _user.sub, _user.name or _user.email, "UPDATE", "firmware", firmware_id, f"{fw.hw_type} v{fw.version} ({fw.channel})" if fw else firmware_id) return fw @router.delete("/{firmware_id}", status_code=204) async def delete_firmware( firmware_id: str, _user: TokenPayload = Depends(require_permission("manufacturing", "delete")), db: AsyncSession = Depends(get_pg_session), ): fw = service.get_firmware(firmware_id) if hasattr(service, "get_firmware") else None service.delete_firmware(firmware_id) label = f"{fw.hw_type} v{fw.version} ({fw.channel})" if fw else firmware_id await log_action(db, _user.sub, _user.name or _user.email, "DELETE", "firmware", firmware_id, label) # ───────────────────────────────────────────────────────────────────────────── # OTA event telemetry — called by devices (no auth, best-effort) # ───────────────────────────────────────────────────────────────────────────── class OtaDownloadEvent(BaseModel): device_uid: str hw_type: str hw_version: str from_version: str to_version: str channel: str class OtaFlashEvent(BaseModel): device_uid: str hw_type: str hw_version: str from_version: str to_version: str channel: str sha256: str @ota_router.post("/events/download", status_code=204) def ota_event_download(event: OtaDownloadEvent): """Device reports that firmware was fully written to flash (pre-commit). No auth required — best-effort telemetry from the device. """ logger.info( "OTA download event: device=%s hw=%s/%s %s → %s (channel=%s)", event.device_uid, event.hw_type, event.hw_version, event.from_version, event.to_version, event.channel, ) service.record_ota_event("download", event.model_dump()) @ota_router.post("/events/flash", status_code=204) def ota_event_flash(event: OtaFlashEvent): """Device reports that firmware partition was committed and device is rebooting. No auth required — best-effort telemetry from the device. """ logger.info( "OTA flash event: device=%s hw=%s/%s %s → %s (channel=%s sha256=%.16s...)", event.device_uid, event.hw_type, event.hw_version, event.from_version, event.to_version, event.channel, event.sha256, ) service.record_ota_event("flash", event.model_dump())