Integration

src/authorizer/globus_auth.py

@@ -0,0 +1,228 @@
+"""
+Models relating to Authorisation for the ESGF Next Gen Core Architecture.
+"""
+
+import logging
+import re
+from typing import Any, Literal
+from urllib.parse import urlparse
+
+from pydantic import BaseModel
+from pydantic_core import ValidationError
+from stac_fastapi.extensions.core.transaction.request import PartialItem
+from stac_pydantic.item import Item
+
+from esgf_core_utils.models.exceptions import (
+    AuthorizationException,
+    MissingPermissionException,
+)
+from esgf_core_utils.models.kafka.events import RequesterData
+
+logger = logging.getLogger("uvicorn.error")
+
+Role = Literal[
+    "CREATE",
+    "UPDATE",
+    "DELETE",
+    "REPLICATE",
+    "REVOKE",
+]
+
+
+class Node(BaseModel):
+    """
+    Model describing Node auth info of a ESGF publisher.
+    """
+
+    id: str
+    roles: set[Role]
+
+
+class Project(BaseModel):
+    """
+    Model describing Project auth info of a ESGF publisher.
+    """
+
+    id: str
+    roles: set[Role]
+
+
+class Nodes(BaseModel):
+    """
+    Model describing Project auth info of a ESGF publisher.
+    """
+
+    nodes: dict[str, Node] = {}
+
+    def add(self, node: Node | dict[str, Any]) -> None:
+        """
+        Add a new project or update roles if project already exists.
+
+        Args:
+            node (Node | dict): node to be added
+        """
+        if isinstance(node, dict):
+            node = Node(**node)
+
+        if existing_node := self.nodes.get(node.id):
+            existing_node.roles.update(node.roles)
+
+        else:
+            self.nodes[node.id] = node
+
+    def authorize_href(self, asset_href: str, role: Role) -> None:
+        asset_url = urlparse(asset_href)
+        node_permission = self.nodes.get(asset_url.hostname or "", None)
+        if not node_permission:
+            node_permission = self.nodes.get("*", None)
+
+        if not node_permission:
+            raise MissingPermissionException(
+                permission_type="node",
+                target=asset_href,
+            )
+
+        if role not in node_permission.roles:
+            raise MissingPermissionException(
+                permission_type="node",
+                role=role,
+                target=asset_href,
+            )
+
+    def authorize(self, assets: dict[str, Any], role: Role) -> None:
+        """Check for appropriate authorisation.
+
+        Args:
+            assets (dict): item to be authorised
+            role (Role): required role for auhroisation
+
+        Raises:
+            MissingPermissionException: Raised if either node or role permission is missing
+        """
+
+        for asset in assets.values():
+            asset = asset.model_dump() if not isinstance(asset, dict) else asset
+
+            if "href" in asset:
+                self.authorize_href(f"https://{asset.get("alternate:name")}", role)
+
+            if alternates := asset.get("alternate"):
+                self.authorize(alternates, role)
+
+
+class Projects(BaseModel):
+    """
+    Model describing Project auth info of a ESGF publisher.
+    """
+
+    projects: dict[str, Project] = {}
+
+    def add(self, project: Project | dict[str, Any]) -> None:
+        """
+        Add a new project or update roles if project already exists.
+
+        Args:
+            project (Project | dict): project to be added
+        """
+        if isinstance(project, dict):
+            project = Project(**project)
+
+        if existing_project := self.projects.get(project.id):
+            existing_project.roles.update(project.roles)
+
+        else:
+            self.projects[project.id] = project
+
+    def authorize(self, project: str, role: Role) -> None:
+        """Check for appropriate authorisation.
+
+        Args:
+            item (Item): item to be authorised
+            role (Role): required role for auhroisation
+
+        Raises:
+            MissingPermissionException: Raised if either node or role permission is missing
+        """
+        project_permission = self.projects.get(project, None)
+        if not project_permission:
+            project_permission = self.projects.get("*", None)
+
+        if not project_permission:
+            raise MissingPermissionException(
+                permission_type="project",
+                target=project,
+            )
+
+        if role not in project_permission.roles:
+            raise MissingPermissionException(
+                permission_type="project",
+                role=role,
+                target=project,
+            )
+
+
+class GlobusAuth(BaseModel):
+    """
+    Model describing Authentication information of a ESGF publisher.
+    """
+
+    requester_data: RequesterData
+    nodes: Nodes = Nodes()
+    projects: Projects = Projects()
+    regex: str
+
+    def authorize(
+        self,
+        collection_id: str,
+        item: Item | PartialItem,
+        role: Role,
+        request_id: str,
+        event_id: str,
+    ) -> None:
+        """Check for appropriate authorisation.
+
+        Args:
+            collection_id: collection id of request
+            item (Item): item to be authorised
+            role (Role): required role for auhroisation
+
+        Raises:
+            AuthorizationException: Raised if either node or role permission is missing
+        """
+        try:
+            self.projects.authorize(collection_id, role)
+            self.nodes.authorize(item.assets or {}, role)
+
+        except MissingPermissionException as exc:
+            raise AuthorizationException(instance=f"{request_id}:{event_id}") from exc
+
+    def add(self, entitlements: list[str]) -> None:
+        """add entitlements to Authorizer.
+
+        Args:
+            entitlements (list[str]): list of entitlements to be added
+        """
+        for entitlement in entitlements:
+            match = re.search(self.regex, entitlement)
+            if match is None:
+                continue
+
+            try:
+                if match.group("type") == "project":
+                    self.projects.add(
+                        Project(
+                            id=match.group("id"),
+                            roles=[match.group("role")],
+                        )
+                    )
+
+                elif match.group("type") == "node":
+                    self.nodes.add(
+                        Node(
+                            id=match.group("id"),
+                            roles=[match.group("role")],
+                        )
+                    )
+
+            except ValidationError:
+                logger.info("Entitlement skipped: %s", entitlement)

src/authorizer/globus_authorizer.py

@@ -4,23 +4,23 @@
 from dataclasses import dataclass
 from threading import Lock
 
+import urllib3
+from esgf_core_utils.models.kafka.events import RequesterData
 from fastapi import Request
 from fastapi.responses import JSONResponse
 from globus_sdk import AccessTokenAuthorizer, GroupsClient
 from globus_sdk.scopes import GroupsScopes
 from starlette.middleware.base import BaseHTTPMiddleware
 
+from authorizer.globus_auth import GlobusAuth
 from settings import settings
 
 logger = logging.getLogger("uvicorn.error")
 
 """
 FastAPI Middleware Authorizer
     Authorizer type: FastAPI Middleware
-    Event payload: Token
     Token source: Authorization
-    Token validation: ^Bearer\\s[^\\s]+$ # noqa: W605
-                      ^Bearer\\s[0-9A-Za-z]+$ for access tokens issued by Globus Auth (?)  # noqa: W605
     Authorization caching: 300 seconds (TRANSACTION_CLIENT__AUTHORIZER_CACHE_TTL_SECONDS)
 """
 
@@ -77,6 +77,80 @@ def _evict_expired(self) -> None:
 _auth_cache = _AuthTTLCache()
 
 
+@dataclass
+class _CachedPolicy:
+    expires_at: float
+    policy: list[str]
+
+
+_policy_cache: _CachedPolicy | None = None
+_policy_lock = Lock()
+
+
+def _load_access_control_policy(policy_path: str) -> list[str]:
+    logger.info("Loading access control policy from %s", policy_path)
+    parsed = urllib3.util.parse_url(policy_path)
+    if parsed.scheme == "file":
+        with open(parsed.path, encoding="utf-8") as file:
+            text = file.read()
+    else:
+        http = urllib3.PoolManager()
+        response = http.request("GET", policy_path)
+        if response.status != 200:
+            raise RuntimeError(f"Failed to load access control policy from {policy_path}: HTTP {response.status}")
+        text = response.data.decode("utf-8")
+
+    lines = [line.strip() for line in text.splitlines() if line.strip()]
+    return lines
+
+
+def get_access_control_policy() -> list[str]:
+    """Return access control policy, reloading from policy_path when cache expires."""
+    global _policy_cache
+
+    ttl = settings.client.policy_cache_ttl_seconds
+    now = time.monotonic()
+
+    with _policy_lock:
+        if _policy_cache is not None and now < _policy_cache.expires_at:
+            return _policy_cache.policy
+        stale_policy = _policy_cache.policy if _policy_cache is not None else None
+
+    try:
+        policy = _load_access_control_policy(settings.client.policy_path)
+    except Exception as exc:
+        if stale_policy is not None:
+            logger.warning("Access control policy refresh failed, using stale cache: %s", exc)
+            return stale_policy
+        raise
+
+    with _policy_lock:
+        _policy_cache = _CachedPolicy(expires_at=now + ttl, policy=policy)
+
+    return policy
+
+
+def _authorizer_context(auth: dict) -> GlobusAuth:
+    """Build GlobusAuth from token auth and cached access control policy entitlements."""
+    token_info = auth["token_info"]
+    user_group_ids = {group["group_id"] for group in auth["groups"]}
+
+    entitlements = [entitlement for entitlement in get_access_control_policy() if entitlement.rsplit(":group:", 1)[-1] in user_group_ids]
+
+    authorizer = GlobusAuth(
+        requester_data=RequesterData(
+            client_id=token_info.get("client_id"),
+            sub=token_info.get("sub"),
+            iss=token_info.get("iss"),
+        ),
+        regex=settings.client.regex,
+    )
+
+    authorizer.add(entitlements)
+
+    return authorizer
+
+
 def _cache_ttl_seconds(token_info: dict, max_ttl: int) -> int:
     exp = token_info.get("exp")
     if exp is None:
@@ -108,7 +182,7 @@ async def dispatch(self, request: Request, call_next):
         access_token = authorization_header[7:].strip()
         cached_auth = _auth_cache.get(access_token)
         if cached_auth is not None:
-            request.state.authorizer = cached_auth
+            request.state.authorizer = _authorizer_context(cached_auth)
             return await call_next(request)
 
         response = settings.client.confidential_client.oauth2_token_introspect(access_token, include="identity_set_detail")
@@ -131,7 +205,7 @@ async def dispatch(self, request: Request, call_next):
         }
         ttl = _cache_ttl_seconds(token_info, settings.client.authorizer_cache_ttl_seconds)
         _auth_cache.set(access_token, auth, ttl)
-        request.state.authorizer = auth
+        request.state.authorizer = _authorizer_context(auth)
         return await call_next(request)
 
     def _validate_token_info(self, token_info: dict) -> JSONResponse | None:
@@ -169,7 +243,6 @@ def get_groups(self, token):
         Amazon API Gateway Authorization caching setting can be use to cache the authorizer response,
         and if the a new request with the same bearer token
         """
-
         tokens = settings.client.confidential_client.oauth2_get_dependent_tokens(token, scope=GroupsScopes.view_my_groups_and_memberships)
         groups_token = tokens.by_resource_server[GroupsClient.resource_server]
         authorizer = AccessTokenAuthorizer(groups_token["access_token"])