Api proxy

This package contains the core code for the API Proxy mechanism.

For more information and context, refer to: - The ADR that introduced this mechanism ⧉ - The technical proposal that describes it ⧉

shared.api_proxy.common

This file contains symbols used by both proxy apps and proxied apps.

For definitions, see the technical proposal here ⧉.

ProxiedEnvName

Bases: AlanBaseEnum

Name of the environments that can be proxied via the API Proxy mechanism, i.e. that can receive calls from other back-ends.

demo class-attribute instance-attribute

demo = 'demo'

local class-attribute instance-attribute

local = 'local'

prod class-attribute instance-attribute

prod = 'prod'

staging class-attribute instance-attribute

staging = 'staging'

ProxyAppName

Bases: AlanBaseEnum

Name of the proxy apps ⧉ in the API Proxy mechanism.

eu_tools class-attribute instance-attribute

eu_tools = 'eu_tools'

fr_api class-attribute instance-attribute

fr_api = 'fr_api'

get_service_account_secret_name_config_key

get_service_account_secret_name_config_key()

Source code in shared/api_proxy/common.py

def get_service_account_secret_name_config_key(self) -> str:
    if self == ProxyAppName.fr_api:
        return "API_PROXY_FR_API_SERVICE_ACCOUNT_SECRET_NAME"
    elif self == ProxyAppName.eu_tools:
        return "API_PROXY_EU_TOOLS_SERVICE_ACCOUNT_SECRET_NAME"
    else:
        raise ValueError(
            f"ProxyAppName does not have an associated service account secret name: {self}"
        )

get_current_proxied_env_name

get_current_proxied_env_name()

Source code in shared/api_proxy/common.py

def get_current_proxied_env_name() -> ProxiedEnvName | None:
    if is_production_mode():
        return ProxiedEnvName.prod
    if is_development_mode():
        return ProxiedEnvName.local
    if is_stage_mode():
        return ProxiedEnvName.staging
    if is_demo_mode():
        return ProxiedEnvName.demo
    return None

shared.api_proxy.proxied

This file contains functions for proxied apps (i.e. another app will access their HTTP functions).

For definitions, see the technical proposal here ⧉.

P module-attribute

P = ParamSpec('P')

ProxiedOnlyBlueprint

ProxiedOnlyBlueprint(
    name,
    import_name,
    permissions_for_endpoints,
    url_prefix=None,
)

Bases: Blueprint

A blueprint for endpoints that can only be reached via the API Proxy mechanism.

This blueprint automatically handles authentication and authorization via the current app's authentication/authorization manager:

• flask_login for EU Tools
• shared.iam for country apps

Constructor for a blueprint that contains proxied-only endpoints.

The permissions_for_endpoints argument is the required permissions for the endpoints that will be registered under this blueprint. A good default value is {EmployeePermission.use_proxied_only_endpoints}, as all proxy apps will have this permission when calling endpoints.

The set acts as an "OR" (more precisely, the principal's permission set must intersect this set).

Source code in shared/api_proxy/proxied.py

def __init__(
    self,
    name: str,
    import_name: str,
    permissions_for_endpoints: set[EmployeePermission],
    url_prefix: str | None = None,
):
    """
    Constructor for a blueprint that contains proxied-only endpoints.

    The `permissions_for_endpoints` argument is the required permissions for the endpoints that will be
    registered under this blueprint. A good default value is `{EmployeePermission.use_proxied_only_endpoints}`, as
    all proxy apps will have this permission when calling endpoints.

    The set acts as an "OR" (more precisely, the principal's permission set must intersect this set).
    """

    super().__init__(name, import_name, url_prefix=url_prefix)

    @self.errorhandler(Exception)
    def _blueprint_error_handler(error: Exception) -> Any:
        return alan_handle_error(error)

    @self.before_request
    def _auth() -> None:
        # HACK @matthieu.stombellini Slightly awkward setup since we don't have access to get_current_app_name within the constructor, as the Flask app isn't ready yet
        if get_current_app_name() == AppName.EU_TOOLS:
            from flask_login import login_required

            @login_required  # type: ignore[misc]
            @permitted_for(permissions_for_endpoints)
            def auth_with_flask_login() -> None:
                pass

            auth_with_flask_login()
        else:
            strategy = GlobalAuthorizationStrategies().alaner_admin(
                permitted_for=permissions_for_endpoints
            )

            @strategy.authenticator()
            def auth_with_shared_iam() -> None:
                with strategy.authorize(
                    Controller=self.__class__,
                    method_name=request.method,
                    call_args={},
                    call_kwargs={},
                ):
                    pass

            auth_with_shared_iam()

T module-attribute

T = TypeVar('T')

shared.api_proxy.proxy

This file contains functions for apps that will act as PROXIES (i.e. they will access other apps' HTTP functions)

KNOWN_PROXIED_APP_NAMES module-attribute

KNOWN_PROXIED_APP_NAMES = keys()

List of names that can be recognized as AppName by the API Proxy system.

Note that this list intentionally contains duplicates (e.g. fr-api and fr_api), use PROXIED_APP_NAMES if you need a list of distinct AppNames for proxied apps.

PROXIED_APP_NAMES module-attribute

PROXIED_APP_NAMES = keys()

List of possible proxied apps in the API Proxy mechanism.

ProxyApiBlueprint

ProxyApiBlueprint(
    proxy_name,
    proxied_url_prefix,
    name,
    import_name,
    url_prefix=None,
    restrict_to_current_env=True,
)

Bases: Blueprint

This blueprint exposes a Proxy HTTP API configured according to the constructor arguments. Can only be used on Proxy Apps (see ProxyAppName).

You MUST secure this blueprint via a before_request hook.

See here ⧉ for more information.

The schema for the API this blueprint exposes (and what endpoint it results in on proxied apps) is described here ⧉

Constructor for ProxyApiBlueprint

Parameters:

Name	Type	Description	Default
proxy_name	ProxyAppName	Name of the proxy app this HTTP API will be exposed from.	required
proxied_url_prefix	str	The prefix that will be appended to requests sent to the proxied app, i.e. {proxied_base_url}/{proxied_url_prefix}/{...path}	required
name	str	The name of this blueprint (passed as-is to the Flask blueprint).	required
import_name	str	The import name of this blueprint (passed as-is to the Flask blueprint).	required
url_prefix	str | None	The URL prefix for THIS schema, which the API will be served under. Defaults to None.	None
restrict_to_current_env	bool	If true (and by default), calls to the proxied app will be done on the same environment as the proxy app. If false, the caller can set which environment they'd like to call as a path parameter. Setting to False is dangerous.	True

Source code in shared/api_proxy/proxy.py

def __init__(
    self,
    proxy_name: ProxyAppName,
    proxied_url_prefix: str,
    name: str,
    import_name: str,
    url_prefix: str | None = None,
    restrict_to_current_env: bool = True,
):
    """Constructor for ProxyApiBlueprint

    Args:
        proxy_name (ProxyAppName): Name of the proxy app this HTTP API will be exposed from.
        proxied_url_prefix (str): The prefix that will be appended to requests sent to the proxied app, i.e. `{proxied_base_url}/{proxied_url_prefix}/{...path}`
        name (str): The name of this blueprint (passed as-is to the Flask blueprint).
        import_name (str): The import name of this blueprint (passed as-is to the Flask blueprint).
        url_prefix (str | None, optional): The URL prefix for THIS schema, which the API will be served under. Defaults to None.
        restrict_to_current_env (bool, optional): If true (and by default), calls to the proxied app will be done on the same environment as the proxy app. If false, the caller can set which environment they'd like to call as a path parameter. **Setting to False is dangerous.**
    """
    super().__init__(name, import_name, url_prefix=url_prefix)

    self._session_builder = ProxyRequestSessionBuilder(
        proxy_name, proxied_url_prefix
    )
    # We want to catch CORS requests in the proxy, not in the proxied API
    CORS(app=self, origins="*", supports_credentials=True)

    if restrict_to_current_env:
        self.add_url_rule(
            "<string:backend_id>/<path:path>",
            view_func=_ProxyApi.as_view(
                "proxy-api",
                session_builder=self._session_builder,
                force_env=get_current_proxied_env_name(),
            ),
        )
    else:
        self.add_url_rule(
            "<string:backend_id>/<string:env>/<path:path>",
            view_func=_ProxyApi.as_view(
                "proxy-api", session_builder=self._session_builder
            ),
        )

    @self.errorhandler(Exception)
    def _blueprint_error_handler(error: Exception) -> Response:
        return alan_handle_error(error)

build_request_session

build_request_session(
    backend_id,
    backend_env=None,
    allow_userless_authentication=False,
)

Source code in shared/api_proxy/proxy.py

def build_request_session(
    self,
    backend_id: str | AppName,
    backend_env: str | ProxiedEnvName | None = None,
    allow_userless_authentication: bool = False,
) -> tuple[str, requests.Session]:
    return self._session_builder.build_request_session(
        backend_id, backend_env, allow_userless_authentication
    )

ProxyRequestSessionBuilder

ProxyRequestSessionBuilder(proxy_name, proxied_url_prefix)

An object that can be used to manually build request sessions to send requests to proxied apps.

Instantiating this class is only useful when you don't have a ProxyApiBlueprint. If you do have a ProxyApiBlueprint, you can use the build_request_session on it directly to create a session.

See here ⧉ for more information on this use case.

Source code in shared/api_proxy/proxy.py

def __init__(self, proxy_name: ProxyAppName, proxied_url_prefix: str):
    self._proxy_name = proxy_name
    self._proxied_url_prefix = proxied_url_prefix

build_request_session

build_request_session(
    backend_id,
    backend_env=None,
    allow_userless_authentication=False,
)

Builds a session for use with the requests library.

Source code in shared/api_proxy/proxy.py

def build_request_session(
    self,
    backend_id: str | AppName,
    backend_env: str | ProxiedEnvName | None = None,
    allow_userless_authentication: bool = False,
) -> tuple[str, requests.Session]:
    """
    Builds a session for use with the `requests` library.
    """
    base_url = mandatory(
        _get_backend_url(
            backend_id,
            backend_env
            or mandatory(
                get_current_proxied_env_name(),
                "Cannot determine current environment",
            ),
        ),
        f"Invalid backend environment: {backend_id}, {backend_env}",
    )

    session = requests.Session()
    session.auth = _ServerToServerAuth(
        base_url=base_url,
        proxy_app_name=self._proxy_name,
        allow_userless_authentication=allow_userless_authentication,
    )
    session.headers = {"X-Request-ID": g.correlation_id}

    return (base_url + self._proxied_url_prefix, session)

shared.api_proxy.utils

create_api_proxy_service_account_users_for_current_env

create_api_proxy_service_account_users_for_current_env(
    commit,
)

Source code in shared/api_proxy/utils.py

def create_api_proxy_service_account_users_for_current_env(commit: bool) -> None:
    from components.global_profile.internal.infrastructure.unit_of_work import (  # noqa: ALN002,ALN043  # Edge case of requiring a "no-local-op" ProfileService
        TransactionUnitOfWork,
    )
    from components.global_profile.public.api import ProfileService  # noqa: ALN002

    current_environment = get_current_proxied_env_name()

    service_accounts_to_create = [
        service_account
        for service_account in _SERVICE_ACCOUNT_TO_APP.keys()
        if current_environment in _SERVICE_ACCOUNT_TO_ENVIRONMENTS[service_account]
    ]

    current_logger.info(f"Will create service accounts {service_accounts_to_create}")

    # TODO: migrate this to use `shared.transaction`
    # For now, we ensure a transaction is open such that anything depending on
    # `shared.transaction` will delegate commit/rollback to us
    # see https://mkdocs.alan.com/shared/transaction/#backward-compatibility

    if not current_session.in_transaction():
        current_session.begin()

    with transaction():
        # Create a simple country-agnostic ProfileService as this is just to create profiles with no other action on local
        # countries
        profile_service = ProfileService(unit_of_work=TransactionUnitOfWork())  # noqa: ALN083

        for service_account in service_accounts_to_create:
            email = _SERVICE_ACCOUNT_TO_EMAIL[service_account]

            profile = profile_service.get_profile_by_email(email)
            if profile is None:
                profile_id = profile_service.create_profile(
                    email=email, first_name=f"Service account {service_account}"
                )
                profile = profile_service.get_or_raise_profile(profile_id)
                current_logger.info(
                    f"Created profile for service account {service_account} on {current_environment}: {profile_id}"
                )

            for app in [
                AppName.ALAN_FR,
                AppName.ALAN_BE,
                AppName.ALAN_CA,
                AppName.ALAN_ES,
            ]:
                if app not in _SERVICE_ACCOUNT_TO_APP[service_account]:
                    continue

                User: type[BaseUser]
                AlanEmployee: type[BaseAlanEmployee]

                if app == AppName.ALAN_FR:
                    from components.fr.internal.models.alan_employee import (  # noqa: ALN069, ALN002
                        AlanEmployee as FrAlanEmployee,
                    )
                    from components.fr.internal.models.user import (  # noqa: ALN069, ALN002
                        User as FrUser,
                    )

                    User = FrUser
                    AlanEmployee = FrAlanEmployee
                elif app == AppName.ALAN_BE:
                    from components.be.internal.models.be_alan_employee import (  # noqa: ALN069, ALN002
                        BeAlanEmployee,
                    )
                    from components.be.internal.models.be_user import (  # noqa: ALN069, ALN002
                        BeUser,
                    )

                    User = BeUser
                    AlanEmployee = BeAlanEmployee

                elif app == AppName.ALAN_ES:
                    from components.es.internal.models.es_alan_employee import (  # noqa: ALN069, ALN002
                        EsAlanEmployee,
                    )
                    from components.es.internal.models.es_user import (  # noqa: ALN069, ALN002
                        EsUser,
                    )

                    User = EsUser
                    AlanEmployee = EsAlanEmployee
                elif app == AppName.ALAN_CA:
                    from components.ca.internal.tech.models.ca_alan_employee import (  # noqa: ALN069, ALN002
                        CaAlanEmployee,
                    )
                    from components.ca.internal.tech.models.ca_user import (  # noqa: ALN069, ALN002
                        CaUser,
                    )

                    User = CaUser
                    AlanEmployee = CaAlanEmployee

                _create_service_account_user(
                    app, User, AlanEmployee, email, service_account, profile.id
                )

    if commit:
        current_session.commit()
    else:
        current_session.flush()

create_eu_tools_api_proxy_service_account_users_for_current_env

create_eu_tools_api_proxy_service_account_users_for_current_env(
    commit,
)

Source code in shared/api_proxy/utils.py

def create_eu_tools_api_proxy_service_account_users_for_current_env(
    commit: bool,
) -> None:
    from apps.eu_tools.alan_home.models.eu_tools_alan_employee import (  # noqa: ALN069, ALN004
        EuToolsAlanEmployee,
    )
    from apps.eu_tools.alan_home.models.service_account import (  # noqa: ALN069, ALN004
        ServiceAccount,
    )

    current_environment = get_current_proxied_env_name()

    service_accounts_to_create = [
        service_account
        for service_account in _SERVICE_ACCOUNT_TO_APP.keys()
        if current_environment in _SERVICE_ACCOUNT_TO_ENVIRONMENTS[service_account]
        and AppName.EU_TOOLS in _SERVICE_ACCOUNT_TO_APP[service_account]
    ]

    # Note: this is a separate function as EU Tools uses a different DB - this needs to be run separately
    for service_account in service_accounts_to_create:
        email = _SERVICE_ACCOUNT_TO_EMAIL[service_account]
        if (
            current_session.query(EuToolsAlanEmployee)  # noqa: ALN085
            .filter(EuToolsAlanEmployee.alan_email == email)
            .count()
            > 0
        ):
            current_logger.info(
                f"Skipping service account {service_account} on EU Tools because a {EuToolsAlanEmployee} already exists with the alan_email {email}"
            )
            continue

        current_logger.info(
            f"Creating EU Tools and Alan Employee for service account {service_account}"
        )
        service_account_model = ServiceAccount(
            # HACK @matthieu.stombellini we'll need to stop hardcoding this when we want to add another ServiceAccount to eu-tools
            id="2111426e-8be7-8020-8808-e7a265de6498",  # ID from Notion: https://www.notion.so/alaninsurance/5984a2e69c9a40e2923c4642e68b0182?v=43e4b5dd0134497cbb2aeb791d29f613&p=2111426e8be780208808e7a265de6498&pm=s
            name=f"API Proxy {service_account}",
            email=email,
            active=True,
            cloudflare_access=False,
        )
        current_session.add(service_account_model)

        alan_employee = EuToolsAlanEmployee(
            alan_email=email,
            service_account=service_account_model,
            roles=[EmployeeRole.api_proxy_service_account],
            start_date=date(2025, 1, 1),
        )
        current_session.add(alan_employee)

    if commit:
        current_session.commit()
    else:
        current_session.flush()