gatehouse-api/gatehouse_app/services/external_auth/models.py

"""External auth models and adapter classes."""
from typing import Optional

from gatehouse_app.extensions import db
from gatehouse_app.models.base import BaseModel
from gatehouse_app.models.auth.authentication_method import (
    ApplicationProviderConfig,
    OrganizationProviderOverride,
)


class ExternalAuthError(Exception):
    """Base exception for external auth errors."""

    def __init__(self, message: str, error_type: str, status_code: int = 400):
        self.message = message
        self.error_type = error_type
        self.status_code = status_code
        super().__init__(message)


class ExternalProviderConfig(BaseModel):
    """OAuth provider configuration per organization.

    DEPRECATED: This model is maintained for backward compatibility only.
    Use ApplicationProviderConfig and OrganizationProviderOverride instead.
    """

    __tablename__ = "external_provider_configs"

    organization_id = db.Column(
        db.String(36), db.ForeignKey("organizations.id"), nullable=False, index=True
    )
    provider_type = db.Column(db.String(50), nullable=False, index=True)
    client_id = db.Column(db.String(255), nullable=False)
    client_secret_encrypted = db.Column(db.String(512), nullable=True)
    auth_url = db.Column(db.String(2048), nullable=False)
    token_url = db.Column(db.String(2048), nullable=False)
    userinfo_url = db.Column(db.String(2048), nullable=True)
    jwks_url = db.Column(db.String(2048), nullable=True)
    scopes = db.Column(db.JSON, nullable=False, default=list)
    redirect_uris = db.Column(db.JSON, nullable=False, default=list)
    settings = db.Column(db.JSON, nullable=True)
    is_active = db.Column(db.Boolean, default=True, nullable=False)

    organization = db.relationship(
        "Organization", back_populates="external_provider_configs"
    )

    __table_args__ = (
        db.Index("idx_provider_config_org", "organization_id", "provider_type"),
        db.UniqueConstraint(
            "organization_id",
            "provider_type",
            name="uix_org_provider_type",
        ),
    )

    def get_client_secret(self) -> str:
        from gatehouse_app.utils.encryption import decrypt
        if self.client_secret_encrypted:
            return decrypt(self.client_secret_encrypted)
        return None

    def set_client_secret(self, secret: str):
        from gatehouse_app.utils.encryption import encrypt
        self.client_secret_encrypted = encrypt(secret)

    def is_redirect_uri_allowed(self, uri: str) -> bool:
        return uri in (self.redirect_uris or [])

    def to_dict(self, include_secrets: bool = False) -> dict:
        data = {
            "id": self.id,
            "organization_id": self.organization_id,
            "provider_type": self.provider_type,
            "client_id": self.client_id,
            "auth_url": self.auth_url,
            "token_url": self.token_url,
            "userinfo_url": self.userinfo_url,
            "scopes": self.scopes,
            "redirect_uris": self.redirect_uris,
            "is_active": self.is_active,
            "settings": self.settings,
            "created_at": self.created_at.isoformat() if self.created_at else None,
            "updated_at": self.updated_at.isoformat() if self.updated_at else None,
        }
        if include_secrets and self.client_secret_encrypted:
            data["client_secret"] = self.get_client_secret()
        return data


class ProviderConfigAdapter:
    """Unified interface for provider configuration.

    Merges application-level config with optional organization overrides.
    """

    def __init__(
        self,
        app_config: ApplicationProviderConfig,
        org_override: Optional[OrganizationProviderOverride] = None,
    ):
        self.app_config = app_config
        self.org_override = org_override
        self.provider_type = app_config.provider_type

    @property
    def client_id(self) -> str:
        if self.org_override and self.org_override.client_id:
            return self.org_override.client_id
        return self.app_config.client_id

    def get_client_secret(self) -> str:
        if self.org_override and self.org_override.client_secret_encrypted:
            return self.org_override.get_client_secret()
        return self.app_config.get_client_secret()

    @property
    def auth_url(self) -> str:
        return self._get_provider_endpoint('auth_url')

    @property
    def token_url(self) -> str:
        return self._get_provider_endpoint('token_url')

    @property
    def userinfo_url(self) -> str:
        return self._get_provider_endpoint('userinfo_url')

    @property
    def jwks_url(self) -> str:
        return self._get_provider_endpoint('jwks_url')

    @property
    def scopes(self) -> list:
        base_scopes = self.app_config.additional_config.get('scopes', []) if self.app_config.additional_config else []
        if self.org_override and self.org_override.additional_config:
            override_scopes = self.org_override.additional_config.get('scopes')
            if override_scopes is not None:
                return override_scopes
        return base_scopes or ['openid', 'profile', 'email']

    @property
    def redirect_uris(self) -> list:
        if self.org_override and self.org_override.redirect_url_override:
            return [self.org_override.redirect_url_override]
        if self.app_config.default_redirect_url:
            return [self.app_config.default_redirect_url]
        return []

    @property
    def settings(self) -> dict:
        settings = {}
        if self.app_config.additional_config:
            settings.update(self.app_config.additional_config)
        if self.org_override and self.org_override.additional_config:
            settings.update(self.org_override.additional_config)
        return settings

    @property
    def is_active(self) -> bool:
        app_enabled = self.app_config.is_enabled
        org_enabled = True if not self.org_override else self.org_override.is_enabled
        return app_enabled and org_enabled

    def is_redirect_uri_allowed(self, uri: str) -> bool:
        return uri in self.redirect_uris

    def _get_provider_endpoint(self, endpoint_name: str) -> Optional[str]:
        if not self.app_config.additional_config:
            return None
        return self.app_config.additional_config.get(endpoint_name)