src.api.http_adapter

  1"""HTTP REST API adapter"""
  2
  3import json
  4from dataclasses import dataclass
  5from typing import override
  6
  7from flask import Flask, Request, request
  8from flask_cors import CORS
  9from prometheus_flask_exporter import (  # pyright: ignore[reportMissingTypeStubs] PrometheusMetrics has no typing
 10    PrometheusMetrics,
 11)
 12
 13from src.config.config_port import ConfigRepoPort
 14from src.offers.offer_service import OfferService, PermissionDeniedError
 15
 16from .api_port import ApiPort
 17
 18
 19class MissingTokenError(Exception):
 20    """Raised when the Authorization header is absent or contains no token."""
 21
 22
 23@dataclass
 24class CreateOfferBody:
 25    """Parsed request body for the create offer endpoint."""
 26
 27    award_id: str
 28
 29
 30@dataclass
 31class Proof:
 32    """Proof object for credential request."""
 33
 34    proof_type: str
 35    jwt: str
 36
 37
 38@dataclass
 39class CredentialRequestBody:
 40    """Parsed request body for the credential endpoint."""
 41
 42    format: str
 43    credential_configuration_id: str
 44    proof: Proof
 45    issuer_state: str
 46
 47
 48class HttpApiAdapter(ApiPort):
 49    """HTTP REST API adapter"""
 50
 51    flask_app: Flask
 52    offer_service: OfferService
 53    config: ConfigRepoPort
 54
 55    def __init__(
 56        self,
 57        config: ConfigRepoPort,
 58        offer_service: OfferService,
 59    ):
 60        """Initialise the adapter.
 61
 62        Args:
 63            config: Application configuration.
 64            metadata_service: Service for credential issuer metadata.
 65            offer_service: Service for creating credential offers.
 66            credential_service: Service for requesting credentials.
 67        """
 68        self.offer_service = offer_service
 69        self.config = config
 70        self.flask_app = self._flask_app()
 71
 72    def _flask_app(self) -> Flask:
 73        app = Flask("HttpApi")
 74
 75        # Configure CORS
 76        allowed_origins = self._parse_allowed_cors_domains()
 77        _ = CORS(app, resources={r"/*": {"origins": allowed_origins}})
 78
 79        # Metrics endpoint is only relevant to HttpAdapter
 80        # no need for service/domain models
 81        metrics: PrometheusMetrics = PrometheusMetrics(app)
 82        _ = metrics.info("app_info", "Application info", version="1.0.3")  # pyright: ignore[reportUnknownMemberType] PrometheusMetrics has no typing
 83
 84        @app.route("/health")
 85        @metrics.do_not_track()
 86        def health() -> str:  # pyright: ignore[reportUnusedFunction] Flask decorators aren't called by design
 87            """Health check endpoint."""
 88            return "OK"
 89
 90        @app.route("/")
 91        def root() -> str:  # pyright: ignore[reportUnusedFunction] Flask decorators aren't called by design
 92            """Root endpoint."""
 93            return "Hello, World!"
 94
 95        @app.route("/api/v1/offers", methods=["POST"])
 96        def create_offer():  # pyright: ignore[reportUnusedFunction] Flask decorators aren't called by design
 97            """Create a credential offer for an achievement."""
 98            try:
 99                bearer_token = self._bearer_token(request)
100            except MissingTokenError:
101                return json.dumps({"error": "Unauthorized"}), 401
102
103            raw: dict[str, str] = request.get_json(silent=True) or {}
104            body = CreateOfferBody(award_id=raw.get("award_id", ""))
105
106            try:
107                offer = self.offer_service.create_offer(
108                    award_id=body.award_id,
109                    bearer_token=bearer_token,
110                )
111            except PermissionDeniedError:
112                return json.dumps({"error": "Forbidden"}), 403
113
114            return json.dumps({"offer_id": offer.offer_id, "uri": offer.uri}), 201
115
116        return app
117
118    def _parse_allowed_cors_domains(self) -> list[str]:
119        """Parse the allowed CORS domains from configuration.
120
121        Returns:
122            List of allowed origins. If the config value is "*", returns ["*"].
123            Otherwise, splits the comma-separated list of domains.
124        """
125        domains = self.config.allowed_cors_domains
126        if domains == "*":
127            return ["*"]
128        return [domain.strip() for domain in domains.split(",")]
129
130    @override
131    def run(self):
132        self.flask_app.run(
133            host=self.config.server_host, port=self.config.server_port, debug=True
134        )
135
136    def _bearer_token(self, request: Request) -> str:
137        """Extract the bearer token from the Authorization header.
138
139        Args:
140            request: The incoming Flask request.
141
142        Returns:
143            The bearer token string.
144
145        Raises:
146            MissingTokenError: When the header is absent or the token is empty.
147        """
148        auth_header = request.authorization
149        if not auth_header or not auth_header.token:
150            raise MissingTokenError()
151        return auth_header.token