Aegis/backend/app/routers/compliance.py

"""Compliance endpoints — framework status, reports, and gap analysis.

Thin HTTP adapter that delegates all data logic to compliance_service.
Provides compliance posture assessment by mapping MITRE ATT&CK technique
coverage to compliance framework controls.
"""

# Import APIRouter, Depends from fastapi
from fastapi import APIRouter, Depends

# Import StreamingResponse from fastapi.responses
from fastapi.responses import StreamingResponse

# Import Session from sqlalchemy.orm
from sqlalchemy.orm import Session

# Import get_db from app.database
from app.database import get_db

# Import get_current_user, require_role from app.dependencies.auth
from app.dependencies.auth import get_current_user, require_role

# Import User from app.models.user
from app.models.user import User

# Import  from app.services.compliance_import_service
from app.services.compliance_import_service import (
    import_cis_controls_v8_mappings,
    import_dora_mappings,
    import_iso_27001_mappings,
    import_iso_42001_mappings,
    import_nist_800_53_mappings,
)

# Import  from app.services.compliance_service
from app.services.compliance_service import (
    build_framework_report_csv,
    get_framework_gaps,
    get_framework_status,
    list_frameworks,
)

# Assign router = APIRouter(prefix="/compliance", tags=["compliance"])
router = APIRouter(prefix="/compliance", tags=["compliance"])


# ── GET /compliance/frameworks ────────────────────────────────────────


@router.get("/frameworks")
# Define function list_frameworks_endpoint
def list_frameworks_endpoint(
    # Entry: db
    db: Session = Depends(get_db),
    # Entry: current_user
    current_user: User = Depends(get_current_user),
) -> list:
    """List all available compliance frameworks.

    Args:
        db (Session): SQLAlchemy database session.
        current_user (User): Authenticated user making the request.

    Returns:
        list: List of framework summary dicts containing id, name, and control counts.
    """
    # Return list_frameworks(db)
    return list_frameworks(db)


# ── GET /compliance/frameworks/{id}/status ────────────────────────────


@router.get("/frameworks/{framework_id}/status")
# Define function framework_status
def framework_status(
    # Entry: framework_id
    framework_id: str,
    # Entry: db
    db: Session = Depends(get_db),
    # Entry: current_user
    current_user: User = Depends(get_current_user),
) -> dict:
    """Get compliance status for each control in a framework.

    Args:
        framework_id (str): Identifier of the compliance framework (e.g. ``nist-800-53``).
        db (Session): SQLAlchemy database session.
        current_user (User): Authenticated user making the request.

    Returns:
        dict: Mapping of control IDs to their coverage status and linked techniques.
    """
    # Return get_framework_status(db, framework_id)
    return get_framework_status(db, framework_id)


# ── GET /compliance/frameworks/{id}/report ────────────────────────────


@router.get("/frameworks/{framework_id}/report")
# Define function framework_report
def framework_report(
    # Entry: framework_id
    framework_id: str,
    # Entry: db
    db: Session = Depends(get_db),
    # Entry: current_user
    current_user: User = Depends(get_current_user),
) -> dict:
    """Get the full compliance report (same as status but marked as report).

    Args:
        framework_id (str): Identifier of the compliance framework.
        db (Session): SQLAlchemy database session.
        current_user (User): Authenticated user making the request.

    Returns:
        dict: Full compliance report with per-control coverage details.
    """
    # Return get_framework_status(db, framework_id)
    return get_framework_status(db, framework_id)


# ── GET /compliance/frameworks/{id}/report/csv ────────────────────────


@router.get("/frameworks/{framework_id}/report/csv")
# Define function framework_report_csv
def framework_report_csv(
    # Entry: framework_id
    framework_id: str,
    # Entry: db
    db: Session = Depends(get_db),
    # Entry: current_user
    current_user: User = Depends(get_current_user),
) -> StreamingResponse:
    """Export compliance report as CSV.

    Args:
        framework_id (str): Identifier of the compliance framework to export.
        db (Session): SQLAlchemy database session.
        current_user (User): Authenticated user making the request.

    Returns:
        StreamingResponse: CSV file attachment with compliance coverage data.
    """
    # csv_bytes, filename = build_framework_report_csv(db, framework_id)
    csv_bytes, filename = build_framework_report_csv(db, framework_id)
    # Return StreamingResponse(
    return StreamingResponse(
        iter([csv_bytes]),
        # Keyword argument: media_type
        media_type="text/csv",
        # Keyword argument: headers
        headers={
            # Literal argument value
            "Content-Disposition": f"attachment; filename={filename}",
        },
    )


# ── GET /compliance/frameworks/{id}/gaps ──────────────────────────────


@router.get("/frameworks/{framework_id}/gaps")
# Define function framework_gaps
def framework_gaps(
    # Entry: framework_id
    framework_id: str,
    # Entry: db
    db: Session = Depends(get_db),
    # Entry: current_user
    current_user: User = Depends(get_current_user),
) -> dict:
    """Get controls with techniques that are not adequately covered.

    Args:
        framework_id (str): Identifier of the compliance framework to analyse.
        db (Session): SQLAlchemy database session.
        current_user (User): Authenticated user making the request.

    Returns:
        dict: Controls flagged as gaps, with linked technique IDs and coverage ratios.
    """
    # Return get_framework_gaps(db, framework_id)
    return get_framework_gaps(db, framework_id)


# ── POST /compliance/import/... ────────────────────────────────────────


@router.post("/import/nist-800-53")
# Define function import_nist
def import_nist(
    # Entry: db
    db: Session = Depends(get_db),
    # Entry: current_user
    current_user: User = Depends(require_role("admin")),
) -> dict:
    """Import NIST 800-53 Rev 5 mappings (admin only).

    Args:
        db (Session): SQLAlchemy database session.
        current_user (User): Authenticated admin user.

    Returns:
        dict: Import result with counts of created and updated control mappings.
    """
    # Assign result = import_nist_800_53_mappings(db)
    result = import_nist_800_53_mappings(db)
    # Return result
    return result


# Apply the @router.post decorator
@router.post("/import/cis-controls-v8")
# Define function import_cis
def import_cis(
    # Entry: db
    db: Session = Depends(get_db),
    # Entry: current_user
    current_user: User = Depends(require_role("admin")),
) -> dict:
    """Import CIS Controls v8 mappings (admin only).

    Args:
        db (Session): SQLAlchemy database session.
        current_user (User): Authenticated admin user.

    Returns:
        dict: Import result with counts of created and updated control mappings.
    """
    # Assign result = import_cis_controls_v8_mappings(db)
    result = import_cis_controls_v8_mappings(db)
    # Return result
    return result


@router.post("/import/dora")
def import_dora(
    db: Session = Depends(get_db),
    current_user: User = Depends(require_role("admin")),
):
    """Import DORA (EU 2022/2554) compliance mappings (admin only)."""
    result = import_dora_mappings(db)
    return result


@router.post("/import/iso-27001")
def import_iso27001(
    db: Session = Depends(get_db),
    current_user: User = Depends(require_role("admin")),
):
    """Import ISO/IEC 27001:2022 Annex A compliance mappings (admin only)."""
    result = import_iso_27001_mappings(db)
    return result


@router.post("/import/iso-42001")
def import_iso42001(
    db: Session = Depends(get_db),
    current_user: User = Depends(require_role("admin")),
):
    """Import ISO/IEC 42001:2023 AI Management System compliance mappings (admin only)."""
    result = import_iso_42001_mappings(db)
    return result