feat(auth): model the token broker — POST /auth/v1/broker/<provider>[/<integration>]

campus_python/auth/v1/__init__.py

@@ -17,6 +17,7 @@
 from ...interface import ResourceRoot
 from ...json_client.interface import JsonClient
 from . import (
+    broker,
     clients,
     connections,
     credentials,
@@ -37,6 +38,7 @@ class AuthRoot(ResourceRoot):
 
     def __init__(self, json_client: JsonClient):
         super().__init__(json_client=json_client)
+        self._broker = None
         self._clients = None
         self._connections = None
         self._credentials = None
@@ -47,6 +49,13 @@ def __init__(self, json_client: JsonClient):
         self._users = None
         self._vaults = None
 
+    @property
+    def broker(self) -> broker.Broker:
+        """Get the token broker resource."""
+        if not self._broker:
+            self._broker = broker.Broker(root=self)
+        return self._broker
+
     @property
     def clients(self) -> clients.Clients:
         """Get the clients resource."""

campus_python/auth/v1/broker.py

@@ -0,0 +1,65 @@
+"""campus.python.auth.v1.broker
+
+Campus Auth token broker resource (v1).
+
+The sanctioned release path for upstream provider access tokens:
+Campus stays sole custodian of upstream credentials, and this endpoint
+hands the short-lived access token (never the refresh token) to
+confidential clients flagged `token_bridge`, on behalf of the bearer's
+user. Errors arrive as APIError subclasses through raise_for_status —
+e.g. 403 carries details.missing_scopes, 401 maps to
+AuthenticationError — so consumers can branch on them without
+hand-mapping statuses.
+"""
+
+from ...interface import JsonDict, ResourceRoot
+
+
+class Broker(ResourceRoot):
+    """Campus Auth token broker resource."""
+    url_prefix = "/auth/v1/broker"
+
+    def __init__(self, root: ResourceRoot):
+        super().__init__(json_client=root.client)
+        self._root = root
+
+    def token(
+            self,
+            provider: str,
+            integration: "str | None" = None,
+            *,
+            min_scopes: "list[str] | None" = None,
+    ) -> JsonDict:
+        """Release the bearer's upstream access token for a provider.
+
+        Args:
+            provider: Base provider (e.g. "google")
+            integration: Integration slug for the namespaced route
+                (e.g. "classroom" → /broker/google/classroom/)
+            min_scopes: Scopes the caller requires; denied unless the
+                user's upstream grant covers them and the client's
+                upstream_scopes allowlist permits them
+
+        Returns:
+            {provider, user_id, access_token, token_type, expires_in,
+            scope} — the refresh token never leaves Campus
+
+        Raises:
+            errors.NotFoundError: No connection for this provider
+                (404); namespaced providers pointed at the identity
+                route are redirected to the integration route the
+                same way
+            errors.AuthenticationError: Bearer session expired (401)
+            errors.APIError: 403 with details.missing_scopes, or 400
+                AUTH_INVALID_SCOPE / configuration problems
+        """
+        if integration:
+            path = self.make_path(f"{provider}/{integration}/")
+        else:
+            path = self.make_path(f"{provider}/")
+        json_body: JsonDict = {}
+        if min_scopes is not None:
+            json_body["min_scopes"] = min_scopes
+        resp = self.client.post(path, json=json_body)
+        resp.raise_for_status()
+        return resp.json()

tests/unit/test_broker.py

@@ -0,0 +1,81 @@
+"""Contract tests for the auth token broker resource (issue #72).
+
+Routes mirror campus/auth/routes/broker.py (campus weekly): POST
+/auth/v1/broker/<provider>/ (identity) and
+/auth/v1/broker/<provider>/<integration>/ (namespaced), body
+{"min_scopes": [...]} or {}. The response is a flat dict carrying the
+upstream access token; errors surface as APIError subclasses via
+raise_for_status (campus-classroom currently raw-requests this
+endpoint and hand-maps the statuses).
+"""
+
+import unittest
+from unittest.mock import Mock
+
+from campus_python.auth.v1 import AuthRoot
+
+
+def make_auth() -> tuple[AuthRoot, Mock]:
+    """Create an AuthRoot backed by a mock JSON client."""
+    client = Mock()
+    return AuthRoot(json_client=client), client
+
+
+BROKER_TOKEN = {
+    "provider": "google",
+    "user_id": "user-1",
+    "access_token": "ya29.upstream",
+    "token_type": "Bearer",
+    "expires_in": 1234,
+    "scope": "email profile",
+}
+
+
+class TestBrokerToken(unittest.TestCase):
+    """broker.token() must POST the identity/integration broker routes."""
+
+    def setUp(self):
+        self.auth, self.client = make_auth()
+        self.client.post.return_value.json.return_value = BROKER_TOKEN
+
+    def test_identity_route_posts_empty_body(self):
+        token = self.auth.broker.token("google")
+        self.client.post.assert_called_once_with(
+            "/auth/v1/broker/google/", json={}
+        )
+        self.assertEqual(token, BROKER_TOKEN)
+
+    def test_min_scopes_passed_through(self):
+        self.auth.broker.token(
+            "google",
+            min_scopes=["https://www.googleapis.com/auth/calendar"],
+        )
+        self.assertEqual(
+            self.client.post.call_args.kwargs["json"],
+            {"min_scopes": ["https://www.googleapis.com/auth/calendar"]},
+        )
+
+    def test_integration_route_targets_namespaced_path(self):
+        self.auth.broker.token("google", "classroom")
+        self.client.post.assert_called_once_with(
+            "/auth/v1/broker/google/classroom/", json={}
+        )
+
+    def test_integration_route_with_min_scopes(self):
+        self.auth.broker.token(
+            "google",
+            "classroom",
+            min_scopes=["https://www.googleapis.com/auth/classroom.rosters"],
+        )
+        self.assertEqual(
+            self.client.post.call_args.kwargs["json"],
+            {
+                "min_scopes": [
+                    "https://www.googleapis.com/auth/classroom.rosters"
+                ]
+            },
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()