import hashlib import hmac import secrets from datetime import UTC, datetime from fastapi import HTTPException, status from sqlalchemy.orm import Session from starlette.requests import Request from app.core.config import settings from app.models.repair import Repair, RepairPublicAccessToken from app.models.user import User from app.repositories.repair_repository import RepairRepository from app.schemas.repair import ( RepairPublicLinkCreateResponse, RepairPublicLinkResponse, RepairPublicStatusHistoryItem, RepairPublicStatusResponse, ) from app.services.audit_service import write_audit_log from app.services.repair_notification_service import STATUS_LABELS from app.services.repair_service import repair_label def public_status_path(token: str) -> str: return f"/status/{token}" class RepairPublicLinkService: @staticmethod def hash_token(token: str) -> str: return hmac.new(settings.secret_key.encode("utf-8"), token.encode("utf-8"), hashlib.sha256).hexdigest() @staticmethod def create_token() -> str: return secrets.token_urlsafe(48) @staticmethod def response_from_link(public_link: RepairPublicAccessToken | None) -> RepairPublicLinkResponse: if public_link is None: return RepairPublicLinkResponse(is_active=False) return RepairPublicLinkResponse( is_active=public_link.is_active, token_hint=public_link.token_hint, expires_at=public_link.expires_at, last_used_at=public_link.last_used_at, created_at=public_link.created_at, revoked_at=public_link.revoked_at, ) @staticmethod def get(db: Session, repair: Repair) -> RepairPublicLinkResponse: return RepairPublicLinkService.response_from_link(RepairRepository.get_active_public_link(db, repair.id)) @staticmethod def create(db: Session, repair: Repair, *, actor: User, request: Request) -> RepairPublicLinkCreateResponse: token = RepairPublicLinkService.create_token() public_link = RepairRepository.create_public_link( db, repair_id=repair.id, token_hash=RepairPublicLinkService.hash_token(token), token_hint=token[-6:], ) write_audit_log( db, action="repairs.public_link.create", entity_type="repairs", entity_id=repair.id, entity_label=repair_label(repair), actor=actor, request=request, metadata={"token_hint": public_link.token_hint}, ) return RepairPublicLinkCreateResponse( is_active=True, token=token, token_hint=public_link.token_hint, expires_at=public_link.expires_at, last_used_at=public_link.last_used_at, created_at=public_link.created_at, revoked_at=public_link.revoked_at, public_status_path=public_status_path(token), ) @staticmethod def revoke(db: Session, repair: Repair, *, actor: User, request: Request) -> RepairPublicLinkResponse: active_link = RepairRepository.get_active_public_link(db, repair.id) RepairRepository.revoke_public_links(db, repair_id=repair.id) write_audit_log( db, action="repairs.public_link.revoke", entity_type="repairs", entity_id=repair.id, entity_label=repair_label(repair), actor=actor, request=request, metadata={"token_hint": active_link.token_hint if active_link else None}, ) return RepairPublicLinkResponse(is_active=False, token_hint=active_link.token_hint if active_link else None, revoked_at=datetime.now(UTC)) @staticmethod def public_status(db: Session, token: str) -> RepairPublicStatusResponse: public_link = RepairRepository.get_public_link_by_hash(db, RepairPublicLinkService.hash_token(token)) if public_link is None or public_link.repair is None: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Reparaturstatus nicht gefunden") RepairRepository.mark_public_link_used(db, public_link) repair = public_link.repair history = RepairRepository.get_history_public(db, repair.id) return RepairPublicStatusResponse( repair_number=repair.repair_number, public_status_label=STATUS_LABELS.get(repair.status, repair.status), device_manufacturer=repair.device_manufacturer, device_model=repair.device_model, status_history_public=[ RepairPublicStatusHistoryItem( status=item.new_status, status_label=STATUS_LABELS.get(item.new_status, item.new_status), created_at=item.created_at, ) for item in history ], updated_at=repair.updated_at, )