Auth (campus.auth)

The /auth/v1 service: OAuth tokens and device flow, login sessions, auth sessions, OAuth clients, user/credential management, vaults, grants, connections, and the token broker. The lanes are described narratively in Authentication.

campus.model.* types referenced here are documented in the campus repo.

AuthRoot (token facade)

campus_python.auth.v1.AuthRoot

Bases: ResourceRoot

Campus Auth resource.

Source code in campus_python/auth/v1/__init__.py

class AuthRoot(ResourceRoot):
    """Campus Auth resource."""
    url_prefix: str = "/auth/v1"

    def __init__(self, json_client: JsonClient):
        super().__init__(json_client=json_client)
        self._broker = None
        self._clients = None
        self._connections = None
        self._credentials = None
        self._grants = None
        self._logins = None
        self._oauth = None
        self._root = None
        self._sessions = None
        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."""
        if not self._clients:
            self._clients = clients.Clients(root=self)
        return self._clients

    @property
    def connections(self) -> connections.Connections:
        """Get the connections resource."""
        if not self._connections:
            self._connections = connections.Connections(root=self)
        return self._connections

    @property
    def credentials(self) -> credentials.Credentials:
        """Get the credentials resource."""
        if not self._credentials:
            self._credentials = credentials.Credentials(root=self)
        return self._credentials

    @property
    def logins(self) -> logins.LoginSessions:
        """Get the logins resource."""
        if not self._logins:
            self._logins = logins.LoginSessions(root=self)
        return self._logins

    @property
    def oauth(self) -> oauth.OAuth:
        """Get the OAuth resource for device authorization flow."""
        if not self._oauth:
            self._oauth = oauth.OAuth(root=self)
        return self._oauth

    @property
    def root(self) -> root.Root:
        """Get the root resource."""
        if not self._root:
            self._root = root.Root(root=self)
        return self._root

    @property
    def sessions(self) -> sessions.CampusSessions:
        """Get the sessions resource."""
        if not self._sessions:
            self._sessions = sessions.CampusSessions(root=self)
        return self._sessions

    @property
    def users(self) -> users.Users:
        """Get the users resource."""
        if not self._users:
            self._users = users.Users(root=self)
        return self._users

    @property
    def grants(self) -> grants.Grants:
        """Get the access-grant store resource."""
        if not self._grants:
            self._grants = grants.Grants(root=self)
        return self._grants

    @property
    def vaults(self) -> vaults.Vaults:
        """Get the vaults resource."""
        if not self._vaults:
            self._vaults = vaults.Vaults(root=self)
        return self._vaults

    def exchange_code_for_token(
            self,
            code: str,
            redirect_uri: str
    ) -> campus.model.OAuthToken:
        """Exchange authorization code for access token.

        Makes POST request to /auth/v1/token which atomically:
        - Validates the authorization code
        - Gets or creates credentials for the user
        - Issues a new token or reuses existing valid token
        - Stores the token in credentials
        - Returns the token

        Note: Credential storage is handled automatically by the token endpoint.

        Args:
            code: The authorization code from OAuth callback
            redirect_uri: The redirect URI used in the auth session

        Returns:
            OAuthToken containing access token and metadata

        Raises:
            AuthenticationError: If code is invalid or token exchange fails
        """
        json_body = {
            "grant_type": "authorization_code",
            "code": code,
            "redirect_uri": redirect_uri,
            "client_id": env.CLIENT_ID,
            "client_secret": env.CLIENT_SECRET,
        }
        token_path = self.url_prefix + "/token"
        return campus.model.OAuthToken.from_resource(
            self._post_json(token_path, json=json_body)
        )

    def get_token(self) -> campus.model.OAuthToken:
        """Convenience method to get access token for a user.

        Re-acquires the login session user's campus bearer via
        POST /logins/<session_id>/token (campus#963/#964, invariant
        A10): one Basic-authenticated endpoint call in the
        session-bound lane — never the operator-gated credentials
        routes (#854). An expired grant token is rotated server-side,
        so the returned token is always live; the response carries no
        refresh token.

        Returns:
            OAuthToken for the user

        Raises:
            AuthenticationError: If no login session is present
            APIError: invalid_grant if the login session or the
                user's campus grant is gone — re-login required

        Example:
            # In a protected view that needs the user's token
            token = campus.auth.get_token()
            headers = {"Authorization": f"Bearer {token.id}"}
            response = requests.get("https://api.example.com/data", headers=headers)
        """
        return self.logins.token()

    def token(
            self,
            grant_type: Literal[
                "client_credentials",
                "refresh_token"
            ],
            *,
            refresh_token: str | None = None,
            client_id: str | None = None,
    ) -> campus.model.OAuthToken:
        """Get OAuth token from the token endpoint.

        Targets the RFC 6749 token endpoint (/auth/v1/oauth/token,
        campus/auth/routes/oauth.py), which serves the device_code,
        refresh_token, and client_credentials grants — NOT the
        authorization-code session endpoint at /auth/v1/token
        (campus/auth/provider.py), whose contract requires code and
        redirect_uri and rejects every other grant type (#60).

        The path passed to the client is relative (like every other
        resource method): CampusRequest._build_url() prepends the
        client's base_url itself, so an absolute URL here would be
        double-prefixed into `https://host/https://host/...` and 404
        against real deployments.

        Args:
            grant_type: "client_credentials" or "refresh_token".
            refresh_token: The refresh token to present (refresh_token
                grant).
            client_id: The OAuth client the grant is made as. The token
                endpoint requires client_id for every grant (campus
                auth/routes/oauth.py token()), including refresh_token;
                defaults to CLIENT_ID from the environment (server
                mode). Public clients without CLIENT_ID configured must
                pass it explicitly (issue #87).
        """
        json_body: dict[str, str] = {
            "grant_type": grant_type,
        }
        match grant_type:
            case "client_credentials":
                json_body["client_id"] = env.CLIENT_ID
                json_body["client_secret"] = env.CLIENT_SECRET
            case "refresh_token":
                if not refresh_token:
                    raise errors.AuthenticationError(
                        error_description="Refresh token required for "
                                          "refresh_token grant type."
                    )
                resolved_client_id = client_id or env.get("CLIENT_ID")
                if not resolved_client_id:
                    raise errors.AuthenticationError(
                        error_description="client_id is required for the "
                                          "refresh_token grant; pass it "
                                          "explicitly or set CLIENT_ID."
                    )
                json_body["client_id"] = resolved_client_id
                json_body["refresh_token"] = refresh_token

        token_path = self.url_prefix + "/oauth/token"
        return campus.model.OAuthToken.from_resource(
            self._post_json(token_path, json=json_body)
        )

    def refresh(
            self,
            stored: campus.model.OAuthToken,
            *,
            client_id: str | None = None,
    ) -> campus.model.OAuthToken:
        """Refresh an OAuth token pair (RFC 6749 section 6).

        Takes a stored token and returns the rotated pair issued by the
        server: a new access token with a new refresh token. The
        presented refresh token is single-use — refresh-token grants
        rotate server-side (campus/auth/routes/oauth.py
        _handle_refresh_token_grant) — so the returned token must be
        persisted before the stale one is presented again.

        This is the public-client entry point (issue #87): it needs no
        client secret, only the client_id the stored token was issued
        to. Error responses (invalid_grant, invalid_client, ...) raise
        APIError subclasses carrying the OAuth error code in details,
        readable via APIError.oauth_error.

        Args:
            stored: The stored OAuthToken whose refresh token is
                presented to the server.
            client_id: The OAuth client the token was issued to;
                defaults to CLIENT_ID from the environment (server
                mode).

        Returns:
            The refreshed OAuthToken (new access + refresh tokens).
        """
        if not stored.refresh_token:
            raise errors.AuthenticationError(
                error_description="Stored token has no refresh token; "
                                  "re-authentication is required."
            )
        return self.token(
            grant_type="refresh_token",
            refresh_token=stored.refresh_token,
            client_id=client_id,
        )

url_prefix: str = '/auth/v1' class-attribute instance-attribute

broker: broker.Broker property

Get the token broker resource.

clients: clients.Clients property

Get the clients resource.

connections: connections.Connections property

Get the connections resource.

credentials: credentials.Credentials property

Get the credentials resource.

logins: logins.LoginSessions property

Get the logins resource.

oauth: oauth.OAuth property

Get the OAuth resource for device authorization flow.

root: root.Root property

Get the root resource.

sessions: sessions.CampusSessions property

Get the sessions resource.

users: users.Users property

Get the users resource.

grants: grants.Grants property

Get the access-grant store resource.

vaults: vaults.Vaults property

Get the vaults resource.

__init__(json_client: JsonClient)

Source code in campus_python/auth/v1/__init__.py

def __init__(self, json_client: JsonClient):
    super().__init__(json_client=json_client)
    self._broker = None
    self._clients = None
    self._connections = None
    self._credentials = None
    self._grants = None
    self._logins = None
    self._oauth = None
    self._root = None
    self._sessions = None
    self._users = None
    self._vaults = None

exchange_code_for_token(code: str, redirect_uri: str) -> campus.model.OAuthToken

Exchange authorization code for access token.

Makes POST request to /auth/v1/token which atomically: - Validates the authorization code - Gets or creates credentials for the user - Issues a new token or reuses existing valid token - Stores the token in credentials - Returns the token

Note: Credential storage is handled automatically by the token endpoint.

Parameters:

Name	Type	Description	Default
code	str	The authorization code from OAuth callback	required
redirect_uri	str	The redirect URI used in the auth session	required

Returns:

Type	Description
OAuthToken	OAuthToken containing access token and metadata

Raises:

Type	Description
AuthenticationError	If code is invalid or token exchange fails

Source code in campus_python/auth/v1/__init__.py

def exchange_code_for_token(
        self,
        code: str,
        redirect_uri: str
) -> campus.model.OAuthToken:
    """Exchange authorization code for access token.

    Makes POST request to /auth/v1/token which atomically:
    - Validates the authorization code
    - Gets or creates credentials for the user
    - Issues a new token or reuses existing valid token
    - Stores the token in credentials
    - Returns the token

    Note: Credential storage is handled automatically by the token endpoint.

    Args:
        code: The authorization code from OAuth callback
        redirect_uri: The redirect URI used in the auth session

    Returns:
        OAuthToken containing access token and metadata

    Raises:
        AuthenticationError: If code is invalid or token exchange fails
    """
    json_body = {
        "grant_type": "authorization_code",
        "code": code,
        "redirect_uri": redirect_uri,
        "client_id": env.CLIENT_ID,
        "client_secret": env.CLIENT_SECRET,
    }
    token_path = self.url_prefix + "/token"
    return campus.model.OAuthToken.from_resource(
        self._post_json(token_path, json=json_body)
    )

get_token() -> campus.model.OAuthToken

Convenience method to get access token for a user.

Re-acquires the login session user's campus bearer via POST /logins//token (campus#963/#964, invariant A10): one Basic-authenticated endpoint call in the session-bound lane — never the operator-gated credentials routes (#854). An expired grant token is rotated server-side, so the returned token is always live; the response carries no refresh token.

Returns:

Type	Description
OAuthToken	OAuthToken for the user

Raises:

Type	Description
AuthenticationError	If no login session is present
APIError	invalid_grant if the login session or the user's campus grant is gone — re-login required

Example

In a protected view that needs the user's token

token = campus.auth.get_token() headers = {"Authorization": f"Bearer {token.id}"} response = requests.get("https://api.example.com/data", headers=headers)

Source code in campus_python/auth/v1/__init__.py

def get_token(self) -> campus.model.OAuthToken:
    """Convenience method to get access token for a user.

    Re-acquires the login session user's campus bearer via
    POST /logins/<session_id>/token (campus#963/#964, invariant
    A10): one Basic-authenticated endpoint call in the
    session-bound lane — never the operator-gated credentials
    routes (#854). An expired grant token is rotated server-side,
    so the returned token is always live; the response carries no
    refresh token.

    Returns:
        OAuthToken for the user

    Raises:
        AuthenticationError: If no login session is present
        APIError: invalid_grant if the login session or the
            user's campus grant is gone — re-login required

    Example:
        # In a protected view that needs the user's token
        token = campus.auth.get_token()
        headers = {"Authorization": f"Bearer {token.id}"}
        response = requests.get("https://api.example.com/data", headers=headers)
    """
    return self.logins.token()

token(grant_type: Literal['client_credentials', 'refresh_token'], *, refresh_token: str | None = None, client_id: str | None = None) -> campus.model.OAuthToken

Get OAuth token from the token endpoint.

Targets the RFC 6749 token endpoint (/auth/v1/oauth/token, campus/auth/routes/oauth.py), which serves the device_code, refresh_token, and client_credentials grants — NOT the authorization-code session endpoint at /auth/v1/token (campus/auth/provider.py), whose contract requires code and redirect_uri and rejects every other grant type (#60).

The path passed to the client is relative (like every other resource method): CampusRequest._build_url() prepends the client's base_url itself, so an absolute URL here would be double-prefixed into https://host/https://host/... and 404 against real deployments.

Parameters:

Name	Type	Description	Default
grant_type	Literal['client_credentials', 'refresh_token']	"client_credentials" or "refresh_token".	required
refresh_token	str | None	The refresh token to present (refresh_token grant).	None
client_id	str | None	The OAuth client the grant is made as. The token endpoint requires client_id for every grant (campus auth/routes/oauth.py token()), including refresh_token; defaults to CLIENT_ID from the environment (server mode). Public clients without CLIENT_ID configured must pass it explicitly (issue #87).	None

Source code in campus_python/auth/v1/__init__.py

def token(
        self,
        grant_type: Literal[
            "client_credentials",
            "refresh_token"
        ],
        *,
        refresh_token: str | None = None,
        client_id: str | None = None,
) -> campus.model.OAuthToken:
    """Get OAuth token from the token endpoint.

    Targets the RFC 6749 token endpoint (/auth/v1/oauth/token,
    campus/auth/routes/oauth.py), which serves the device_code,
    refresh_token, and client_credentials grants — NOT the
    authorization-code session endpoint at /auth/v1/token
    (campus/auth/provider.py), whose contract requires code and
    redirect_uri and rejects every other grant type (#60).

    The path passed to the client is relative (like every other
    resource method): CampusRequest._build_url() prepends the
    client's base_url itself, so an absolute URL here would be
    double-prefixed into `https://host/https://host/...` and 404
    against real deployments.

    Args:
        grant_type: "client_credentials" or "refresh_token".
        refresh_token: The refresh token to present (refresh_token
            grant).
        client_id: The OAuth client the grant is made as. The token
            endpoint requires client_id for every grant (campus
            auth/routes/oauth.py token()), including refresh_token;
            defaults to CLIENT_ID from the environment (server
            mode). Public clients without CLIENT_ID configured must
            pass it explicitly (issue #87).
    """
    json_body: dict[str, str] = {
        "grant_type": grant_type,
    }
    match grant_type:
        case "client_credentials":
            json_body["client_id"] = env.CLIENT_ID
            json_body["client_secret"] = env.CLIENT_SECRET
        case "refresh_token":
            if not refresh_token:
                raise errors.AuthenticationError(
                    error_description="Refresh token required for "
                                      "refresh_token grant type."
                )
            resolved_client_id = client_id or env.get("CLIENT_ID")
            if not resolved_client_id:
                raise errors.AuthenticationError(
                    error_description="client_id is required for the "
                                      "refresh_token grant; pass it "
                                      "explicitly or set CLIENT_ID."
                )
            json_body["client_id"] = resolved_client_id
            json_body["refresh_token"] = refresh_token

    token_path = self.url_prefix + "/oauth/token"
    return campus.model.OAuthToken.from_resource(
        self._post_json(token_path, json=json_body)
    )

refresh(stored: campus.model.OAuthToken, *, client_id: str | None = None) -> campus.model.OAuthToken

Refresh an OAuth token pair (RFC 6749 section 6).

Takes a stored token and returns the rotated pair issued by the server: a new access token with a new refresh token. The presented refresh token is single-use — refresh-token grants rotate server-side (campus/auth/routes/oauth.py _handle_refresh_token_grant) — so the returned token must be persisted before the stale one is presented again.

This is the public-client entry point (issue #87): it needs no client secret, only the client_id the stored token was issued to. Error responses (invalid_grant, invalid_client, ...) raise APIError subclasses carrying the OAuth error code in details, readable via APIError.oauth_error.

Parameters:

Name	Type	Description	Default
stored	OAuthToken	The stored OAuthToken whose refresh token is presented to the server.	required
client_id	str | None	The OAuth client the token was issued to; defaults to CLIENT_ID from the environment (server mode).	None

Returns:

Type	Description
OAuthToken	The refreshed OAuthToken (new access + refresh tokens).

Source code in campus_python/auth/v1/__init__.py

def refresh(
        self,
        stored: campus.model.OAuthToken,
        *,
        client_id: str | None = None,
) -> campus.model.OAuthToken:
    """Refresh an OAuth token pair (RFC 6749 section 6).

    Takes a stored token and returns the rotated pair issued by the
    server: a new access token with a new refresh token. The
    presented refresh token is single-use — refresh-token grants
    rotate server-side (campus/auth/routes/oauth.py
    _handle_refresh_token_grant) — so the returned token must be
    persisted before the stale one is presented again.

    This is the public-client entry point (issue #87): it needs no
    client secret, only the client_id the stored token was issued
    to. Error responses (invalid_grant, invalid_client, ...) raise
    APIError subclasses carrying the OAuth error code in details,
    readable via APIError.oauth_error.

    Args:
        stored: The stored OAuthToken whose refresh token is
            presented to the server.
        client_id: The OAuth client the token was issued to;
            defaults to CLIENT_ID from the environment (server
            mode).

    Returns:
        The refreshed OAuthToken (new access + refresh tokens).
    """
    if not stored.refresh_token:
        raise errors.AuthenticationError(
            error_description="Stored token has no refresh token; "
                              "re-authentication is required."
        )
    return self.token(
        grant_type="refresh_token",
        refresh_token=stored.refresh_token,
        client_id=client_id,
    )

OAuth (device flow, revoke)

campus_python.auth.v1.oauth.OAuth

Bases: ResourceRoot

OAuth 2.0 Device Authorization Flow resource.

Provides methods for device authorization flow used by CLI and other applications with limited input capabilities.

Reference: https://datatracker.ietf.org/doc/html/rfc8628

Source code in campus_python/auth/v1/oauth.py

class OAuth(ResourceRoot):
    """OAuth 2.0 Device Authorization Flow resource.

    Provides methods for device authorization flow used by CLI
    and other applications with limited input capabilities.

    Reference: https://datatracker.ietf.org/doc/html/rfc8628
    """

    # All OAuth endpoints live under /auth/v1/oauth (campus/auth/routes/
    # oauth.py); the auth app enables strict_slashes, so the slash-less
    # relative paths built here must match the route rules exactly.
    url_prefix = "/auth/v1/oauth"

    def __init__(self, root: ResourceRoot):
        super().__init__(json_client=root.client)
        self._root = root

    def request_device_code(
            self,
            client_id: str,
    ) -> dict:
        """Request a device code for the device authorization flow.

        Args:
            client_id: The OAuth client ID (e.g., "campus-cli")

        Returns:
            Dict containing:
            - device_code: The device code for polling
            - user_code: The code the user must enter
            - verification_uri: The URI where the user enters the code
            - verification_uri_complete: The URI with user_code pre-filled
            - expires_in: Seconds until the device code expires
            - interval: Minimum seconds between polling attempts

        Raises:
            AuthenticationError: If client_id is invalid
        """
        json_body = {
            "client_id": client_id,
        }
        return self._post_json(self.make_path("device_authorize"), json=json_body)

    def poll_for_token(
            self,
            client_id: str,
            device_code: str,
    ) -> dict:
        """Poll the token endpoint for an access token.

        Args:
            client_id: The OAuth client ID
            device_code: The device code received from request_device_code()

        Returns:
            Dict containing:
            - access_token: The access token
            - token_type: Token type (usually "Bearer")
            - expires_in: Seconds until token expires
            - refresh_token: The refresh token
            - scope: Granted scopes

        Raises:
            AuthenticationError: With specific error codes:
            - authorization_pending: User hasn't completed auth yet
            - slow_down: Client is polling too fast
            - expired_token: Device code has expired
            - access_denied: User denied the authorization
        """
        json_body = {
            "grant_type": "urn:ietf:params:oauth:grant-type:device_code",
            "client_id": client_id,
            "device_code": device_code,
        }
        resp = self.client.post(self.make_path("token"), json=json_body)

        # Handle OAuth error responses
        if resp.status_code == 400:
            oauth_error, message = _parse_oauth_error(resp.json())

            # Map RFC 8628 errors to AuthenticationError; the OAuth error
            # code travels in details so callers can read it back via the
            # APIError.oauth_error property.
            descriptions = {
                "authorization_pending": "Authorization pending",
                "slow_down": "Slow down",
                "expired_token": "Device code has expired",
                "access_denied": "Access denied by user",
            }
            raise errors.AuthenticationError(
                status_code=400,
                error_description=(
                    descriptions.get(oauth_error or "")
                    or message
                    or "Unknown error"
                ),
                details={"oauth_error": oauth_error},
            )

        resp.raise_for_status()
        return resp.json()

    def wait_for_token(
            self,
            client_id: str,
            device_code: str,
            *,
            interval: int | None = None,
            max_attempts: int = 60,
            on_pending: Callable[[], None] | None = None,
            sleep: Callable[[float], None] = time.sleep,
    ) -> dict:
        """Poll the token endpoint until the device is authorized.

        Implements the polling loop RFC 8628 §3.5 clients must run on
        top of poll_for_token(): authorization_pending keeps polling,
        slow_down raises the interval by SLOW_DOWN_ADJUSTMENT seconds
        with the raised interval persisting for the remainder of the
        flow, network failures are retried on non-final attempts, and
        every other error (expired_token, access_denied, unknown) is
        fatal. See the parity table in this module's docstring.

        Args:
            client_id: The OAuth client ID (e.g., "campus-cli")
            device_code: The device code from request_device_code()
            interval: Minimum seconds between poll attempts, from the
                request_device_code() response. Defaults to
                DEFAULT_POLL_INTERVAL when absent or zero.
            max_attempts: Maximum number of poll attempts; callers
                typically derive this from the device code's expires_in.
            on_pending: Invoked after each authorization_pending response,
                for progress reporting (e.g. printing a dot per poll).
            sleep: The sleep function; injectable for tests.

        Returns:
            The token response dict (access_token, refresh_token, ...),
            as returned by poll_for_token().

        Raises:
            AuthenticationError: On fatal OAuth errors or when
                max_attempts is exhausted without authorization.
            errors.ServerError: When the final attempt fails at the
                network/5xx level.
        """
        poll_interval = (
            max(1, int(interval)) if interval else DEFAULT_POLL_INTERVAL
        )
        last_attempt = max_attempts - 1

        for attempt in range(max_attempts):
            try:
                return self.poll_for_token(
                    client_id=client_id, device_code=device_code
                )
            except errors.AuthenticationError as err:
                if err.oauth_error == "authorization_pending":
                    if on_pending is not None:
                        on_pending()
                    sleep(poll_interval)
                elif err.oauth_error == "slow_down":
                    # RFC 8628 §3.5: the raised interval persists for
                    # the remainder of the flow, not just the next
                    # attempt.
                    poll_interval += SLOW_DOWN_ADJUSTMENT
                    sleep(poll_interval)
                else:
                    raise
            except errors.ServerError:
                # Network failure: retry on non-final attempts only
                if attempt == last_attempt:
                    raise
                sleep(poll_interval)

        raise errors.AuthenticationError(
            error_description=(
                "Device authorization timed out after "
                f"{max_attempts} poll attempts; restart the flow."
            )
        )

    def authorize_device(
            self,
            user_code: str,
            user_id: str,
    ) -> dict:
        """Authorize a device code with a user ID.

        This is called by the verification page when a user submits
        their user code.

        Args:
            user_code: The user code from the CLI
            user_id: The user ID of the authorizing user

        Returns:
            Dict with success status

        Raises:
            NotFoundError: If user_code is invalid or expired
            ConflictError: If user_code is already authorized/denied
        """
        json_body = {
            "user_code": user_code,
            "user_id": user_id,
        }
        return self._post_json(
            self.make_path("device/authorize"), json=json_body
        )

    def revoke(
            self,
            token: str,
            client_id: str,
            token_type_hint: "str | None" = None,
    ) -> None:
        """Revoke an access or refresh token (RFC 7009).

        Args:
            token: The access or refresh token to revoke
            client_id: The OAuth client the token was issued to
            token_type_hint: Optional "access_token" or "refresh_token"

        Per RFC 7009 section 2.2 the server returns 200 regardless of
        whether the token was found or already revoked, so callers
        cannot use this endpoint to confirm a token's validity.
        """
        self._post_json(self.make_path("revoke"), json=payload(
            token=token,
            client_id=client_id,
            token_type_hint=token_type_hint,
        ))
        return None

url_prefix = '/auth/v1/oauth' class-attribute instance-attribute

__init__(root: ResourceRoot)

Source code in campus_python/auth/v1/oauth.py

def __init__(self, root: ResourceRoot):
    super().__init__(json_client=root.client)
    self._root = root

request_device_code(client_id: str) -> dict

Request a device code for the device authorization flow.

Parameters:

Name	Type	Description	Default
client_id	str	The OAuth client ID (e.g., "campus-cli")	required

Returns:

Type	Description
dict	Dict containing:
dict	device_code: The device code for polling
dict	user_code: The code the user must enter
dict	verification_uri: The URI where the user enters the code
dict	verification_uri_complete: The URI with user_code pre-filled
dict	expires_in: Seconds until the device code expires
dict	interval: Minimum seconds between polling attempts

Raises:

Type	Description
AuthenticationError	If client_id is invalid

Source code in campus_python/auth/v1/oauth.py

def request_device_code(
        self,
        client_id: str,
) -> dict:
    """Request a device code for the device authorization flow.

    Args:
        client_id: The OAuth client ID (e.g., "campus-cli")

    Returns:
        Dict containing:
        - device_code: The device code for polling
        - user_code: The code the user must enter
        - verification_uri: The URI where the user enters the code
        - verification_uri_complete: The URI with user_code pre-filled
        - expires_in: Seconds until the device code expires
        - interval: Minimum seconds between polling attempts

    Raises:
        AuthenticationError: If client_id is invalid
    """
    json_body = {
        "client_id": client_id,
    }
    return self._post_json(self.make_path("device_authorize"), json=json_body)

poll_for_token(client_id: str, device_code: str) -> dict

Poll the token endpoint for an access token.

Parameters:

Name	Type	Description	Default
client_id	str	The OAuth client ID	required
device_code	str	The device code received from request_device_code()	required

Returns:

Type	Description
dict	Dict containing:
dict	access_token: The access token
dict	token_type: Token type (usually "Bearer")
dict	expires_in: Seconds until token expires
dict	refresh_token: The refresh token
dict	scope: Granted scopes

Raises:

Type	Description
AuthenticationError	With specific error codes:
-authorization_pending	User hasn't completed auth yet
-slow_down	Client is polling too fast
-expired_token	Device code has expired
-access_denied	User denied the authorization

Source code in campus_python/auth/v1/oauth.py

def poll_for_token(
        self,
        client_id: str,
        device_code: str,
) -> dict:
    """Poll the token endpoint for an access token.

    Args:
        client_id: The OAuth client ID
        device_code: The device code received from request_device_code()

    Returns:
        Dict containing:
        - access_token: The access token
        - token_type: Token type (usually "Bearer")
        - expires_in: Seconds until token expires
        - refresh_token: The refresh token
        - scope: Granted scopes

    Raises:
        AuthenticationError: With specific error codes:
        - authorization_pending: User hasn't completed auth yet
        - slow_down: Client is polling too fast
        - expired_token: Device code has expired
        - access_denied: User denied the authorization
    """
    json_body = {
        "grant_type": "urn:ietf:params:oauth:grant-type:device_code",
        "client_id": client_id,
        "device_code": device_code,
    }
    resp = self.client.post(self.make_path("token"), json=json_body)

    # Handle OAuth error responses
    if resp.status_code == 400:
        oauth_error, message = _parse_oauth_error(resp.json())

        # Map RFC 8628 errors to AuthenticationError; the OAuth error
        # code travels in details so callers can read it back via the
        # APIError.oauth_error property.
        descriptions = {
            "authorization_pending": "Authorization pending",
            "slow_down": "Slow down",
            "expired_token": "Device code has expired",
            "access_denied": "Access denied by user",
        }
        raise errors.AuthenticationError(
            status_code=400,
            error_description=(
                descriptions.get(oauth_error or "")
                or message
                or "Unknown error"
            ),
            details={"oauth_error": oauth_error},
        )

    resp.raise_for_status()
    return resp.json()

wait_for_token(client_id: str, device_code: str, *, interval: int | None = None, max_attempts: int = 60, on_pending: Callable[[], None] | None = None, sleep: Callable[[float], None] = time.sleep) -> dict

Poll the token endpoint until the device is authorized.

Implements the polling loop RFC 8628 §3.5 clients must run on top of poll_for_token(): authorization_pending keeps polling, slow_down raises the interval by SLOW_DOWN_ADJUSTMENT seconds with the raised interval persisting for the remainder of the flow, network failures are retried on non-final attempts, and every other error (expired_token, access_denied, unknown) is fatal. See the parity table in this module's docstring.

Parameters:

Name	Type	Description	Default
client_id	str	The OAuth client ID (e.g., "campus-cli")	required
device_code	str	The device code from request_device_code()	required
interval	int | None	Minimum seconds between poll attempts, from the request_device_code() response. Defaults to DEFAULT_POLL_INTERVAL when absent or zero.	None
max_attempts	int	Maximum number of poll attempts; callers typically derive this from the device code's expires_in.	60
on_pending	Callable[[], None] | None	Invoked after each authorization_pending response, for progress reporting (e.g. printing a dot per poll).	None
sleep	Callable[[float], None]	The sleep function; injectable for tests.	sleep

Returns:

Type	Description
dict	The token response dict (access_token, refresh_token, ...),
dict	as returned by poll_for_token().

Raises:

Type	Description
AuthenticationError	On fatal OAuth errors or when max_attempts is exhausted without authorization.
ServerError	When the final attempt fails at the network/5xx level.

Source code in campus_python/auth/v1/oauth.py

def wait_for_token(
        self,
        client_id: str,
        device_code: str,
        *,
        interval: int | None = None,
        max_attempts: int = 60,
        on_pending: Callable[[], None] | None = None,
        sleep: Callable[[float], None] = time.sleep,
) -> dict:
    """Poll the token endpoint until the device is authorized.

    Implements the polling loop RFC 8628 §3.5 clients must run on
    top of poll_for_token(): authorization_pending keeps polling,
    slow_down raises the interval by SLOW_DOWN_ADJUSTMENT seconds
    with the raised interval persisting for the remainder of the
    flow, network failures are retried on non-final attempts, and
    every other error (expired_token, access_denied, unknown) is
    fatal. See the parity table in this module's docstring.

    Args:
        client_id: The OAuth client ID (e.g., "campus-cli")
        device_code: The device code from request_device_code()
        interval: Minimum seconds between poll attempts, from the
            request_device_code() response. Defaults to
            DEFAULT_POLL_INTERVAL when absent or zero.
        max_attempts: Maximum number of poll attempts; callers
            typically derive this from the device code's expires_in.
        on_pending: Invoked after each authorization_pending response,
            for progress reporting (e.g. printing a dot per poll).
        sleep: The sleep function; injectable for tests.

    Returns:
        The token response dict (access_token, refresh_token, ...),
        as returned by poll_for_token().

    Raises:
        AuthenticationError: On fatal OAuth errors or when
            max_attempts is exhausted without authorization.
        errors.ServerError: When the final attempt fails at the
            network/5xx level.
    """
    poll_interval = (
        max(1, int(interval)) if interval else DEFAULT_POLL_INTERVAL
    )
    last_attempt = max_attempts - 1

    for attempt in range(max_attempts):
        try:
            return self.poll_for_token(
                client_id=client_id, device_code=device_code
            )
        except errors.AuthenticationError as err:
            if err.oauth_error == "authorization_pending":
                if on_pending is not None:
                    on_pending()
                sleep(poll_interval)
            elif err.oauth_error == "slow_down":
                # RFC 8628 §3.5: the raised interval persists for
                # the remainder of the flow, not just the next
                # attempt.
                poll_interval += SLOW_DOWN_ADJUSTMENT
                sleep(poll_interval)
            else:
                raise
        except errors.ServerError:
            # Network failure: retry on non-final attempts only
            if attempt == last_attempt:
                raise
            sleep(poll_interval)

    raise errors.AuthenticationError(
        error_description=(
            "Device authorization timed out after "
            f"{max_attempts} poll attempts; restart the flow."
        )
    )

authorize_device(user_code: str, user_id: str) -> dict

Authorize a device code with a user ID.

This is called by the verification page when a user submits their user code.

Parameters:

Name	Type	Description	Default
user_code	str	The user code from the CLI	required
user_id	str	The user ID of the authorizing user	required

Returns:

Type	Description
dict	Dict with success status

Raises:

Type	Description
NotFoundError	If user_code is invalid or expired
ConflictError	If user_code is already authorized/denied

Source code in campus_python/auth/v1/oauth.py

def authorize_device(
        self,
        user_code: str,
        user_id: str,
) -> dict:
    """Authorize a device code with a user ID.

    This is called by the verification page when a user submits
    their user code.

    Args:
        user_code: The user code from the CLI
        user_id: The user ID of the authorizing user

    Returns:
        Dict with success status

    Raises:
        NotFoundError: If user_code is invalid or expired
        ConflictError: If user_code is already authorized/denied
    """
    json_body = {
        "user_code": user_code,
        "user_id": user_id,
    }
    return self._post_json(
        self.make_path("device/authorize"), json=json_body
    )

revoke(token: str, client_id: str, token_type_hint: str | None = None) -> None

Revoke an access or refresh token (RFC 7009).

Parameters:

Name	Type	Description	Default
token	str	The access or refresh token to revoke	required
client_id	str	The OAuth client the token was issued to	required
token_type_hint	str | None	Optional "access_token" or "refresh_token"	None

Per RFC 7009 section 2.2 the server returns 200 regardless of whether the token was found or already revoked, so callers cannot use this endpoint to confirm a token's validity.

Source code in campus_python/auth/v1/oauth.py

def revoke(
        self,
        token: str,
        client_id: str,
        token_type_hint: "str | None" = None,
) -> None:
    """Revoke an access or refresh token (RFC 7009).

    Args:
        token: The access or refresh token to revoke
        client_id: The OAuth client the token was issued to
        token_type_hint: Optional "access_token" or "refresh_token"

    Per RFC 7009 section 2.2 the server returns 200 regardless of
    whether the token was found or already revoked, so callers
    cannot use this endpoint to confirm a token's validity.
    """
    self._post_json(self.make_path("revoke"), json=payload(
        token=token,
        client_id=client_id,
        token_type_hint=token_type_hint,
    ))
    return None

LoginSessions

campus_python.auth.v1.logins.LoginSessions

Bases: ResourceCollection

Campus Auth Logins resource.

Source code in campus_python/auth/v1/logins.py

class LoginSessions(ResourceCollection):
    """Campus Auth Logins resource."""
    path = "logins"

    def __getitem__(
            self,
            session_id: str | None = None
    ) -> "LoginSessions.Login":
        """Get a login session resource by ID.

        Without an explicit ID, the id resolves from the installed
        session-id provider. There is no framework fallback: hosts
        without a provider must pass the id explicitly.
        """
        session_id = session_id or sessionstore.resolve_session_id(
            sessionstore.LANE_LOGIN
        )
        if not session_id:
            raise errors.AuthenticationError(
                error_description=(
                    "No login session ID available: pass session_id or "
                    "install a session-id provider "
                    "(campus_python.auth.v1.set_session_id_provider)."
                )
            )
        return LoginSessions.Login(session_id, parent=self)

    def from_session(
            self,
            session_id: str | None = None,
    ) -> campus.model.LoginSession:
        """Get a login session using a client-side session ID.

        Resolution order: explicit session_id > the installed
        session-id provider. There is no framework fallback.
        """
        return self[session_id].get()

    def has_session(self) -> bool:
        """True when the installed session-id provider supplies a login
        id for this lane."""
        return bool(sessionstore.resolve_session_id(sessionstore.LANE_LOGIN))

    def token(self, session_id: str | None = None) -> campus.model.OAuthToken:
        """Re-acquire the current login session's campus bearer (#117).

        One endpoint call (campus#964, invariant A10):
        POST /logins/<session_id>/token, authenticated by the client's
        Basic credentials plus the session id — user bearers are denied
        outright, so the caller must not present one (revoke_session()
        first mid-rotation). An expired grant token is rotated
        server-side; the response omits the refresh token.

        The session id resolves explicitly > installed provider.

        Raises:
            AuthenticationError: If no login session is present.
            APIError: invalid_grant when the session or the user's
                campus grant is gone — the cue to require re-login.
        """
        return self[session_id].token()

    def new(
            self,
            *,
            user_id: str,
            device_id: str | None = None,
            agent_string: str,
    ) -> campus.model.LoginSession:
        """Create a new login session.

        Mirrors the payload expected by the logins routes. Storing the
        returned session's id client-side is the host's business
        (flask_campus context owns its flask.session storage).
        """
        client_id = env.CLIENT_ID
        return campus.model.LoginSession.from_resource(self._post_json(
            self.make_path(),
            json=payload(
                client_id=str(client_id),
                user_id=str(user_id),
                agent_string=agent_string,
                device_id=device_id,
            )
        ))

    class Login(Resource):
        """A single login session resource."""

        @property
        def session_id(self) -> str:
            """The login session id embedded in this resource's path."""
            return self.path.rstrip("/").split("/")[-1]

        def revoke(self) -> None:
            """Revoke this login session (log the user out)."""
            self._delete_json(self.make_path(end_slash=True))

        def get(self) -> campus.model.LoginSession:
            """Get this login session.

            Returns:
                The LoginSession.
            """
            body = self._get_json(self.make_path(end_slash=True))
            return campus.model.LoginSession.from_resource(body)

        def update(self, *, expiry_seconds: int) -> campus.model.LoginSession:
            """Extend or shorten this login session.

            Args:
                expiry_seconds: New session lifetime in seconds.

            Returns:
                The updated LoginSession.
            """
            body = self._patch_json(
                self.make_path(end_slash=True),
                json={"expiry_seconds": expiry_seconds}
            )
            return campus.model.LoginSession.from_resource(body)

        def token(self) -> campus.model.OAuthToken:
            """Re-acquire the session user's campus bearer (#117).

            POST /logins/<session_id>/token — the session-bound
            re-acquisition lane (campus#964, invariant A10): the login
            session plays the authorization code's role, authenticated
            with the client's Basic credentials. Issuance mirrors the
            code exchange: an unexpired grant token is reused, an
            expired one is minted fresh (rotated server-side), and a
            missing grant is an invalid_grant, never minted from
            nothing (C5). The response omits the refresh token —
            campus refresh material never leaves the auth service in
            this lane.

            The call presents Basic client credentials, never the
            ambient bearer: the route denies user principals outright
            (#963/A10), and with_user_session() nested inside another
            with_user_session() (e.g. an auth-broker release reached
            from a route that already holds the user session) would
            otherwise present the outer session's user token and be
            refused with 403. Any prior Authorization header is
            restored afterwards, so an enclosing session keeps its
            authorization until it re-runs use_token() with the fresh
            bearer.
            """
            # Canonical path has no trailing slash (route rule is
            # /<session_id>/token, unlike the CRUD subresources).
            prior_authorization = self.client.authorization
            self.client.reset_authorization()
            try:
                body = self._post_json(self.make_path("token"))
            finally:
                if prior_authorization is not None and prior_authorization.startswith(
                        "Bearer "):
                    self.client.set_bearer_authorization(
                        prior_authorization[len("Bearer "):]
                    )
            return campus.model.OAuthToken.from_resource(body)

path = 'logins' class-attribute instance-attribute

Login

Bases: Resource

A single login session resource.

Source code in campus_python/auth/v1/logins.py

class Login(Resource):
    """A single login session resource."""

    @property
    def session_id(self) -> str:
        """The login session id embedded in this resource's path."""
        return self.path.rstrip("/").split("/")[-1]

    def revoke(self) -> None:
        """Revoke this login session (log the user out)."""
        self._delete_json(self.make_path(end_slash=True))

    def get(self) -> campus.model.LoginSession:
        """Get this login session.

        Returns:
            The LoginSession.
        """
        body = self._get_json(self.make_path(end_slash=True))
        return campus.model.LoginSession.from_resource(body)

    def update(self, *, expiry_seconds: int) -> campus.model.LoginSession:
        """Extend or shorten this login session.

        Args:
            expiry_seconds: New session lifetime in seconds.

        Returns:
            The updated LoginSession.
        """
        body = self._patch_json(
            self.make_path(end_slash=True),
            json={"expiry_seconds": expiry_seconds}
        )
        return campus.model.LoginSession.from_resource(body)

    def token(self) -> campus.model.OAuthToken:
        """Re-acquire the session user's campus bearer (#117).

        POST /logins/<session_id>/token — the session-bound
        re-acquisition lane (campus#964, invariant A10): the login
        session plays the authorization code's role, authenticated
        with the client's Basic credentials. Issuance mirrors the
        code exchange: an unexpired grant token is reused, an
        expired one is minted fresh (rotated server-side), and a
        missing grant is an invalid_grant, never minted from
        nothing (C5). The response omits the refresh token —
        campus refresh material never leaves the auth service in
        this lane.

        The call presents Basic client credentials, never the
        ambient bearer: the route denies user principals outright
        (#963/A10), and with_user_session() nested inside another
        with_user_session() (e.g. an auth-broker release reached
        from a route that already holds the user session) would
        otherwise present the outer session's user token and be
        refused with 403. Any prior Authorization header is
        restored afterwards, so an enclosing session keeps its
        authorization until it re-runs use_token() with the fresh
        bearer.
        """
        # Canonical path has no trailing slash (route rule is
        # /<session_id>/token, unlike the CRUD subresources).
        prior_authorization = self.client.authorization
        self.client.reset_authorization()
        try:
            body = self._post_json(self.make_path("token"))
        finally:
            if prior_authorization is not None and prior_authorization.startswith(
                    "Bearer "):
                self.client.set_bearer_authorization(
                    prior_authorization[len("Bearer "):]
                )
        return campus.model.OAuthToken.from_resource(body)

session_id: str property

The login session id embedded in this resource's path.

revoke() -> None

Revoke this login session (log the user out).

Source code in campus_python/auth/v1/logins.py

def revoke(self) -> None:
    """Revoke this login session (log the user out)."""
    self._delete_json(self.make_path(end_slash=True))

get() -> campus.model.LoginSession

Get this login session.

Returns:

Type	Description
LoginSession	The LoginSession.

Source code in campus_python/auth/v1/logins.py

def get(self) -> campus.model.LoginSession:
    """Get this login session.

    Returns:
        The LoginSession.
    """
    body = self._get_json(self.make_path(end_slash=True))
    return campus.model.LoginSession.from_resource(body)

update(*, expiry_seconds: int) -> campus.model.LoginSession

Extend or shorten this login session.

Parameters:

Name	Type	Description	Default
expiry_seconds	int	New session lifetime in seconds.	required

Returns:

Type	Description
LoginSession	The updated LoginSession.

Source code in campus_python/auth/v1/logins.py

def update(self, *, expiry_seconds: int) -> campus.model.LoginSession:
    """Extend or shorten this login session.

    Args:
        expiry_seconds: New session lifetime in seconds.

    Returns:
        The updated LoginSession.
    """
    body = self._patch_json(
        self.make_path(end_slash=True),
        json={"expiry_seconds": expiry_seconds}
    )
    return campus.model.LoginSession.from_resource(body)

token() -> campus.model.OAuthToken

Re-acquire the session user's campus bearer (#117).

POST /logins//token — the session-bound re-acquisition lane (campus#964, invariant A10): the login session plays the authorization code's role, authenticated with the client's Basic credentials. Issuance mirrors the code exchange: an unexpired grant token is reused, an expired one is minted fresh (rotated server-side), and a missing grant is an invalid_grant, never minted from nothing (C5). The response omits the refresh token — campus refresh material never leaves the auth service in this lane.

The call presents Basic client credentials, never the ambient bearer: the route denies user principals outright (#963/A10), and with_user_session() nested inside another with_user_session() (e.g. an auth-broker release reached from a route that already holds the user session) would otherwise present the outer session's user token and be refused with 403. Any prior Authorization header is restored afterwards, so an enclosing session keeps its authorization until it re-runs use_token() with the fresh bearer.

Source code in campus_python/auth/v1/logins.py

def token(self) -> campus.model.OAuthToken:
    """Re-acquire the session user's campus bearer (#117).

    POST /logins/<session_id>/token — the session-bound
    re-acquisition lane (campus#964, invariant A10): the login
    session plays the authorization code's role, authenticated
    with the client's Basic credentials. Issuance mirrors the
    code exchange: an unexpired grant token is reused, an
    expired one is minted fresh (rotated server-side), and a
    missing grant is an invalid_grant, never minted from
    nothing (C5). The response omits the refresh token —
    campus refresh material never leaves the auth service in
    this lane.

    The call presents Basic client credentials, never the
    ambient bearer: the route denies user principals outright
    (#963/A10), and with_user_session() nested inside another
    with_user_session() (e.g. an auth-broker release reached
    from a route that already holds the user session) would
    otherwise present the outer session's user token and be
    refused with 403. Any prior Authorization header is
    restored afterwards, so an enclosing session keeps its
    authorization until it re-runs use_token() with the fresh
    bearer.
    """
    # Canonical path has no trailing slash (route rule is
    # /<session_id>/token, unlike the CRUD subresources).
    prior_authorization = self.client.authorization
    self.client.reset_authorization()
    try:
        body = self._post_json(self.make_path("token"))
    finally:
        if prior_authorization is not None and prior_authorization.startswith(
                "Bearer "):
            self.client.set_bearer_authorization(
                prior_authorization[len("Bearer "):]
            )
    return campus.model.OAuthToken.from_resource(body)

__getitem__(session_id: str | None = None) -> LoginSessions.Login

Get a login session resource by ID.

Without an explicit ID, the id resolves from the installed session-id provider. There is no framework fallback: hosts without a provider must pass the id explicitly.

Source code in campus_python/auth/v1/logins.py

def __getitem__(
        self,
        session_id: str | None = None
) -> "LoginSessions.Login":
    """Get a login session resource by ID.

    Without an explicit ID, the id resolves from the installed
    session-id provider. There is no framework fallback: hosts
    without a provider must pass the id explicitly.
    """
    session_id = session_id or sessionstore.resolve_session_id(
        sessionstore.LANE_LOGIN
    )
    if not session_id:
        raise errors.AuthenticationError(
            error_description=(
                "No login session ID available: pass session_id or "
                "install a session-id provider "
                "(campus_python.auth.v1.set_session_id_provider)."
            )
        )
    return LoginSessions.Login(session_id, parent=self)

from_session(session_id: str | None = None) -> campus.model.LoginSession

Get a login session using a client-side session ID.

Resolution order: explicit session_id > the installed session-id provider. There is no framework fallback.

Source code in campus_python/auth/v1/logins.py

def from_session(
        self,
        session_id: str | None = None,
) -> campus.model.LoginSession:
    """Get a login session using a client-side session ID.

    Resolution order: explicit session_id > the installed
    session-id provider. There is no framework fallback.
    """
    return self[session_id].get()

has_session() -> bool

True when the installed session-id provider supplies a login id for this lane.

Source code in campus_python/auth/v1/logins.py

def has_session(self) -> bool:
    """True when the installed session-id provider supplies a login
    id for this lane."""
    return bool(sessionstore.resolve_session_id(sessionstore.LANE_LOGIN))

token(session_id: str | None = None) -> campus.model.OAuthToken

Re-acquire the current login session's campus bearer (#117).

One endpoint call (campus#964, invariant A10): POST /logins//token, authenticated by the client's Basic credentials plus the session id — user bearers are denied outright, so the caller must not present one (revoke_session() first mid-rotation). An expired grant token is rotated server-side; the response omits the refresh token.

The session id resolves explicitly > installed provider.

Raises:

Type	Description
AuthenticationError	If no login session is present.
APIError	invalid_grant when the session or the user's campus grant is gone — the cue to require re-login.

Source code in campus_python/auth/v1/logins.py

def token(self, session_id: str | None = None) -> campus.model.OAuthToken:
    """Re-acquire the current login session's campus bearer (#117).

    One endpoint call (campus#964, invariant A10):
    POST /logins/<session_id>/token, authenticated by the client's
    Basic credentials plus the session id — user bearers are denied
    outright, so the caller must not present one (revoke_session()
    first mid-rotation). An expired grant token is rotated
    server-side; the response omits the refresh token.

    The session id resolves explicitly > installed provider.

    Raises:
        AuthenticationError: If no login session is present.
        APIError: invalid_grant when the session or the user's
            campus grant is gone — the cue to require re-login.
    """
    return self[session_id].token()

new(*, user_id: str, device_id: str | None = None, agent_string: str) -> campus.model.LoginSession

Create a new login session.

Mirrors the payload expected by the logins routes. Storing the returned session's id client-side is the host's business (flask_campus context owns its flask.session storage).

Source code in campus_python/auth/v1/logins.py

def new(
        self,
        *,
        user_id: str,
        device_id: str | None = None,
        agent_string: str,
) -> campus.model.LoginSession:
    """Create a new login session.

    Mirrors the payload expected by the logins routes. Storing the
    returned session's id client-side is the host's business
    (flask_campus context owns its flask.session storage).
    """
    client_id = env.CLIENT_ID
    return campus.model.LoginSession.from_resource(self._post_json(
        self.make_path(),
        json=payload(
            client_id=str(client_id),
            user_id=str(user_id),
            agent_string=agent_string,
            device_id=device_id,
        )
    ))

CampusSessions

campus_python.auth.v1.sessions.CampusSessions

Bases: ResourceCollection

Campus Auth Sessions resource.

Source code in campus_python/auth/v1/sessions.py

class CampusSessions(ResourceCollection):
    """Campus Auth Sessions resource."""
    path = f"sessions/{PROVIDER}"

    def __getitem__(self, session_id: str) -> "CampusSessions.Session":
        return CampusSessions.Session(session_id, parent=self)

    def from_code(self, code: str) -> campus.model.AuthSession:
        """Get a session using authorization code."""
        return campus.model.AuthSession.from_resource(self._post_json(
            self.make_path("authorization_code", end_slash=False),
            json={"code": code}
        ))

    def from_session(self, session_id: str | None = None) -> campus.model.AuthSession:
        """Get a session by its client-side session id.

        Resolution order: explicit session_id > the installed
        session-id provider. There is no framework fallback: hosts
        without a provider must pass the id explicitly.
        """
        if session_id is None:
            session_id = sessionstore.resolve_session_id(
                sessionstore.LANE_SESSION
            )
        if not session_id:
            raise errors.AuthenticationError(
                error_description=(
                    "No session id available: pass session_id or "
                    "install a session-id provider "
                    "(campus_python.auth.v1.set_session_id_provider)."
                )
            )
        return CampusSessions.Session(session_id, parent=self).get()

    def has_session(self) -> bool:
        """True when the installed session-id provider supplies a
        session id for this lane."""
        return bool(sessionstore.resolve_session_id(sessionstore.LANE_SESSION))

    def new(
            self,
            *,
            # expiry_seconds: int,
            user_id: str | None = None,
            redirect_uri: str,
            scopes: "list[str]",
            target: str | None = None,
    ) -> campus.model.AuthSession:
        """Create a new auth session (start the OAuth authorize flow).

        Args:
            user_id: Optional user to pre-authorize.
            redirect_uri: OAuth redirect URI.
            scopes: Scopes to request.
            target: Where to redirect after authorize.

        Returns:
            The created AuthSession.
        """
        client_id = env.CLIENT_ID
        json_data: JsonDict = {
            # "expiry_seconds": expiry_seconds,
            "client_id": client_id,
            "redirect_uri": redirect_uri,
            "scopes": scopes,
        }
        if target is not None:
            json_data["target"] = target
        if user_id is not None:
            json_data["user_id"] = str(user_id)
        # The returned session id doubles as the OAuth state parameter;
        # storing it client-side is the host's business (flask_campus
        # context owns its flask.session storage).
        return campus.model.AuthSession.from_resource(
            self._post_json(self.make_path(), json=json_data)
        )

    def list(
            self,
            *,
            provider: str | None = PROVIDER,
            client_id: str | None = None,
            user_id: str | None = None,
            limit: int | None = None,
            offset: int | None = None,
    ) -> "list[campus.model.AuthSession]":
        """List live sessions (admin view, campus#1055).

        GET /sessions/ with optional provider/client_id/user_id
        filters and limit/offset pagination. Requires the operator
        principal or a designated admin user (sessions:read); client
        principals are denied. Records carry metadata only — no
        authorization_code, no user embeds — and finalized sessions
        are absent (finalize deletes the record).
        """
        query: JsonDict = {}
        if provider is not None:
            query["provider"] = provider
        if client_id is not None:
            query["client_id"] = client_id
        if user_id is not None:
            query["user_id"] = str(user_id)
        if limit is not None:
            query["limit"] = limit
        if offset is not None:
            query["offset"] = offset
        # The list route is provider-agnostic (/sessions/), unlike the
        # campus-provider path this collection wraps.
        collection_path = f"/{self.root.make_path('sessions').strip('/')}/"
        body = self._get_json(collection_path, query=query)
        return [
            campus.model.AuthSession.from_resource(item)
            for item in body.get("sessions", [])
        ]

    def sweep(self, at_time: datetime | str | None = None) -> int:
        """Finalize expired auth sessions.

        Args:
            at_time: Cutoff time (defaults to now, UTC).

        Returns:
            Number of sessions swept.
        """
        json_data: JsonDict = {}
        match at_time:
            case datetime():
                json_data["at_time"] = (
                    schema.DateTime.from_datetime(at_time)
                )
            case str():
                json_data["at_time"] = schema.DateTime(at_time)
            case None:
                json_data["at_time"] = schema.DateTime.utcnow()
        body = self._post_json(self.make_path("sweep", end_slash=False), json=json_data)
        return int(body["swept_count"])

    class Session(Resource):
        """A single provider session (/sessions/campus/{session_id})."""

        @property
        def session_id(self) -> str:
            """The auth session id embedded in this resource's path."""
            return self.path.rstrip("/").split("/")[-1]

        def finalize(self) -> "tuple[str, campus.model.User | None]":
            """Finalize this session (post-authorization).

            Returns:
                (target_url, user_or_None): where to redirect the
                browser, plus the session user when the calling client
                owns the session; None from pre-#879 services.
            """
            # DELETE /sessions/{provider}/{session_id} ->
            # {"target": <url>, "user": {...}|null}
            body = self._delete_json(self.make_path(end_slash=True))
            # The auth service embeds the session user for the owning
            # client (#879); None from pre-#879 services.
            user = campus.model.User.from_resource(body["user"]) \
                if body.get("user") else None
            return body["target"], user

        def get(self) -> campus.model.AuthSession:
            """Get this auth session.

            Returns:
                The AuthSession.
            """
            body = self._get_json(self.make_path(end_slash=True))
            return campus.model.AuthSession.from_resource(body)

        def update(self, **updates) -> None:
            """Update this auth session.

            Only user_id and authorization_code are expected by the
            API; keyword arguments map to session fields.
            """
            # Only user_id and authorization_code are expected by the API
            self._patch_json(self.make_path(end_slash=True), json=updates)
            return None

path = f'sessions/{PROVIDER}' class-attribute instance-attribute

Session

Bases: Resource

A single provider session (/sessions/campus/{session_id}).

Source code in campus_python/auth/v1/sessions.py

class Session(Resource):
    """A single provider session (/sessions/campus/{session_id})."""

    @property
    def session_id(self) -> str:
        """The auth session id embedded in this resource's path."""
        return self.path.rstrip("/").split("/")[-1]

    def finalize(self) -> "tuple[str, campus.model.User | None]":
        """Finalize this session (post-authorization).

        Returns:
            (target_url, user_or_None): where to redirect the
            browser, plus the session user when the calling client
            owns the session; None from pre-#879 services.
        """
        # DELETE /sessions/{provider}/{session_id} ->
        # {"target": <url>, "user": {...}|null}
        body = self._delete_json(self.make_path(end_slash=True))
        # The auth service embeds the session user for the owning
        # client (#879); None from pre-#879 services.
        user = campus.model.User.from_resource(body["user"]) \
            if body.get("user") else None
        return body["target"], user

    def get(self) -> campus.model.AuthSession:
        """Get this auth session.

        Returns:
            The AuthSession.
        """
        body = self._get_json(self.make_path(end_slash=True))
        return campus.model.AuthSession.from_resource(body)

    def update(self, **updates) -> None:
        """Update this auth session.

        Only user_id and authorization_code are expected by the
        API; keyword arguments map to session fields.
        """
        # Only user_id and authorization_code are expected by the API
        self._patch_json(self.make_path(end_slash=True), json=updates)
        return None

session_id: str property

The auth session id embedded in this resource's path.

finalize() -> tuple[str, campus.model.User | None]

Finalize this session (post-authorization).

Returns:

Type	Description
(target_url, user_or_None)	where to redirect the
User | None	browser, plus the session user when the calling client
tuple[str, User | None]	owns the session; None from pre-#879 services.

Source code in campus_python/auth/v1/sessions.py

def finalize(self) -> "tuple[str, campus.model.User | None]":
    """Finalize this session (post-authorization).

    Returns:
        (target_url, user_or_None): where to redirect the
        browser, plus the session user when the calling client
        owns the session; None from pre-#879 services.
    """
    # DELETE /sessions/{provider}/{session_id} ->
    # {"target": <url>, "user": {...}|null}
    body = self._delete_json(self.make_path(end_slash=True))
    # The auth service embeds the session user for the owning
    # client (#879); None from pre-#879 services.
    user = campus.model.User.from_resource(body["user"]) \
        if body.get("user") else None
    return body["target"], user

get() -> campus.model.AuthSession

Get this auth session.

Returns:

Type	Description
AuthSession	The AuthSession.

Source code in campus_python/auth/v1/sessions.py

def get(self) -> campus.model.AuthSession:
    """Get this auth session.

    Returns:
        The AuthSession.
    """
    body = self._get_json(self.make_path(end_slash=True))
    return campus.model.AuthSession.from_resource(body)

update(**updates) -> None

Update this auth session.

Only user_id and authorization_code are expected by the API; keyword arguments map to session fields.

Source code in campus_python/auth/v1/sessions.py

def update(self, **updates) -> None:
    """Update this auth session.

    Only user_id and authorization_code are expected by the
    API; keyword arguments map to session fields.
    """
    # Only user_id and authorization_code are expected by the API
    self._patch_json(self.make_path(end_slash=True), json=updates)
    return None

__getitem__(session_id: str) -> CampusSessions.Session

Source code in campus_python/auth/v1/sessions.py

def __getitem__(self, session_id: str) -> "CampusSessions.Session":
    return CampusSessions.Session(session_id, parent=self)

from_code(code: str) -> campus.model.AuthSession

Get a session using authorization code.

Source code in campus_python/auth/v1/sessions.py

def from_code(self, code: str) -> campus.model.AuthSession:
    """Get a session using authorization code."""
    return campus.model.AuthSession.from_resource(self._post_json(
        self.make_path("authorization_code", end_slash=False),
        json={"code": code}
    ))

from_session(session_id: str | None = None) -> campus.model.AuthSession

Get a session by its client-side session id.

Resolution order: explicit session_id > the installed session-id provider. There is no framework fallback: hosts without a provider must pass the id explicitly.

Source code in campus_python/auth/v1/sessions.py

def from_session(self, session_id: str | None = None) -> campus.model.AuthSession:
    """Get a session by its client-side session id.

    Resolution order: explicit session_id > the installed
    session-id provider. There is no framework fallback: hosts
    without a provider must pass the id explicitly.
    """
    if session_id is None:
        session_id = sessionstore.resolve_session_id(
            sessionstore.LANE_SESSION
        )
    if not session_id:
        raise errors.AuthenticationError(
            error_description=(
                "No session id available: pass session_id or "
                "install a session-id provider "
                "(campus_python.auth.v1.set_session_id_provider)."
            )
        )
    return CampusSessions.Session(session_id, parent=self).get()

has_session() -> bool

True when the installed session-id provider supplies a session id for this lane.

Source code in campus_python/auth/v1/sessions.py

def has_session(self) -> bool:
    """True when the installed session-id provider supplies a
    session id for this lane."""
    return bool(sessionstore.resolve_session_id(sessionstore.LANE_SESSION))

new(*, user_id: str | None = None, redirect_uri: str, scopes: list[str], target: str | None = None) -> campus.model.AuthSession

Create a new auth session (start the OAuth authorize flow).

Parameters:

Name	Type	Description	Default
user_id	str | None	Optional user to pre-authorize.	None
redirect_uri	str	OAuth redirect URI.	required
scopes	list[str]	Scopes to request.	required
target	str | None	Where to redirect after authorize.	None

Returns:

Type	Description
AuthSession	The created AuthSession.

Source code in campus_python/auth/v1/sessions.py

def new(
        self,
        *,
        # expiry_seconds: int,
        user_id: str | None = None,
        redirect_uri: str,
        scopes: "list[str]",
        target: str | None = None,
) -> campus.model.AuthSession:
    """Create a new auth session (start the OAuth authorize flow).

    Args:
        user_id: Optional user to pre-authorize.
        redirect_uri: OAuth redirect URI.
        scopes: Scopes to request.
        target: Where to redirect after authorize.

    Returns:
        The created AuthSession.
    """
    client_id = env.CLIENT_ID
    json_data: JsonDict = {
        # "expiry_seconds": expiry_seconds,
        "client_id": client_id,
        "redirect_uri": redirect_uri,
        "scopes": scopes,
    }
    if target is not None:
        json_data["target"] = target
    if user_id is not None:
        json_data["user_id"] = str(user_id)
    # The returned session id doubles as the OAuth state parameter;
    # storing it client-side is the host's business (flask_campus
    # context owns its flask.session storage).
    return campus.model.AuthSession.from_resource(
        self._post_json(self.make_path(), json=json_data)
    )

list(*, provider: str | None = PROVIDER, client_id: str | None = None, user_id: str | None = None, limit: int | None = None, offset: int | None = None) -> list[campus.model.AuthSession]

List live sessions (admin view, campus#1055).

GET /sessions/ with optional provider/client_id/user_id filters and limit/offset pagination. Requires the operator principal or a designated admin user (sessions:read); client principals are denied. Records carry metadata only — no authorization_code, no user embeds — and finalized sessions are absent (finalize deletes the record).

Source code in campus_python/auth/v1/sessions.py

def list(
        self,
        *,
        provider: str | None = PROVIDER,
        client_id: str | None = None,
        user_id: str | None = None,
        limit: int | None = None,
        offset: int | None = None,
) -> "list[campus.model.AuthSession]":
    """List live sessions (admin view, campus#1055).

    GET /sessions/ with optional provider/client_id/user_id
    filters and limit/offset pagination. Requires the operator
    principal or a designated admin user (sessions:read); client
    principals are denied. Records carry metadata only — no
    authorization_code, no user embeds — and finalized sessions
    are absent (finalize deletes the record).
    """
    query: JsonDict = {}
    if provider is not None:
        query["provider"] = provider
    if client_id is not None:
        query["client_id"] = client_id
    if user_id is not None:
        query["user_id"] = str(user_id)
    if limit is not None:
        query["limit"] = limit
    if offset is not None:
        query["offset"] = offset
    # The list route is provider-agnostic (/sessions/), unlike the
    # campus-provider path this collection wraps.
    collection_path = f"/{self.root.make_path('sessions').strip('/')}/"
    body = self._get_json(collection_path, query=query)
    return [
        campus.model.AuthSession.from_resource(item)
        for item in body.get("sessions", [])
    ]

sweep(at_time: datetime | str | None = None) -> int

Finalize expired auth sessions.

Parameters:

Name	Type	Description	Default
at_time	datetime | str | None	Cutoff time (defaults to now, UTC).	None

Returns:

Type	Description
int	Number of sessions swept.

Source code in campus_python/auth/v1/sessions.py

def sweep(self, at_time: datetime | str | None = None) -> int:
    """Finalize expired auth sessions.

    Args:
        at_time: Cutoff time (defaults to now, UTC).

    Returns:
        Number of sessions swept.
    """
    json_data: JsonDict = {}
    match at_time:
        case datetime():
            json_data["at_time"] = (
                schema.DateTime.from_datetime(at_time)
            )
        case str():
            json_data["at_time"] = schema.DateTime(at_time)
        case None:
            json_data["at_time"] = schema.DateTime.utcnow()
    body = self._post_json(self.make_path("sweep", end_slash=False), json=json_data)
    return int(body["swept_count"])

Clients

campus_python.auth.v1.clients.Clients

Bases: ResourceCollection

Campus Auth Clients resource.

Source code in campus_python/auth/v1/clients.py

class Clients(ResourceCollection):
    """Campus Auth Clients resource."""
    path = "clients"

    def __getitem__(self, client_id: str) -> "Clients.Client":
        """Get a specific client resource by ID."""
        return Clients.Client(client_id, parent=self)

    def list(self) -> "list[campus.model.Client]":
        """List all OAuth clients.

        Returns:
            List of Client objects.
        """
        body = self._get_json(self.make_path())
        return [
            campus.model.Client.from_resource(item)
            for item in body["clients"]
        ]

    def new(
            self,
            name: str,
            description: str,
            is_public: bool = False,
            redirect_uris: list[str] | None = None,
            allowed_scopes: list[str] | None = None,
            upstream_scopes: dict[str, list[str]] | None = None,
            token_bridge: bool | None = None
    ) -> campus.model.Client:
        """Create a new client.

        Args:
            name: Client name
            description: Client description
            is_public: True for public clients (CLI/mobile apps) without secrets
            redirect_uris: OAuth redirect URIs for public clients
            allowed_scopes: Token-scope allowlist (fail-closed: an
                empty list grants nothing)
            upstream_scopes: Per-provider upstream scope map
                (e.g. {"google": [...]})
            token_bridge: Token bridge access flag (upstream token
                release; rejected by the server for public clients)

        Returns:
            The created Client
        """
        return campus.model.Client.from_resource(self._post_json(
            self.make_path(),
            json=payload(
                name=name,
                description=description,
                is_public=is_public,
                redirect_uris=redirect_uris or [],
                allowed_scopes=allowed_scopes,
                upstream_scopes=upstream_scopes,
                token_bridge=token_bridge,
            )
        ))

    class Client(Resource):
        """Campus Auth Client resource."""

        @property
        def access(self) -> "Clients.Client.ClientAccess":
            return Clients.Client.ClientAccess("access", parent=self)

        @property
        def client_id(self) -> str:
            """The client_id of the resource was passed in as a path
            parameter, and is extracted using this property alias.
            """
            return self.path.strip("/")

        def delete(self) -> None:
            """Delete this client."""
            self._delete_json(self.make_path())

        def get(self) -> campus.model.Client:
            """Get this client.

            Returns:
                The Client.
            """
            body = self._get_json(self.make_path())
            return campus.model.Client.from_resource(body)

        def revoke(self) -> str:
            """Revoke the client secret and return the new one.

            Returns:
                The newly generated client secret.
            """
            return self._post_json(self.make_path("revoke", end_slash=False))["secret"]

        def update(
                self,
                name: str | None = None,
                description: str | None = None,
                redirect_uris: list[str] | None = None,
                allowed_scopes: list[str] | None = None,
                upstream_scopes: dict[str, list[str]] | None = None,
                token_bridge: bool | None = None
        ) -> campus.model.Client:
            """Update the client.

            Only the fields passed are sent; the server replaces each
            provided field wholesale (PATCH semantics), so list and
            dict values must carry the full desired contents.

            Args:
                name: New client name
                description: New client description
                redirect_uris: New OAuth redirect URIs
                allowed_scopes: Replacement token-scope allowlist
                    (fail-closed: an empty list grants nothing)
                upstream_scopes: Replacement per-provider upstream
                    scope map (e.g. {"google": [...]})
                token_bridge: Token bridge access flag (rejected by the
                    server for public clients)

            Returns:
                The updated Client
            """
            return campus.model.Client.from_resource(self._patch_json(
                self.make_path(),
                json=payload(
                    name=name,
                    description=description,
                    redirect_uris=redirect_uris,
                    allowed_scopes=allowed_scopes,
                    upstream_scopes=upstream_scopes,
                    token_bridge=token_bridge,
                )
            ))

        class ClientAccess(Resource):
            """Campus Auth Client Access resource."""

            def get(
                    self,
                    vault: str | None = None
            ) -> JsonDict:
                """Get this client's vault access mapping.

                Args:
                    vault: Scope the mapping to one vault label.

                Returns:
                    {vault_label: permission} for the accessible vaults.
                """
                # GET rule is /<client_id>/access/ (trailing slash) and the
                # auth app sets strict_slashes — the slash is required.
                if vault:
                    return self._get_json(
                        self.make_path(end_slash=True),
                        query={"vault": vault}
                    )
                return self._get_json(self.make_path(end_slash=True))

            def check(
                    self,
                    vault: str,
                    permission: int,
            ) -> JsonDict:
                """Check whether the client holds a permission on a vault.

                GET /<client_id>/access/check?vault=&permission=
                Returns: {"vault": ..., "permission": <bool>}
                """
                return self._get_json(
                    self.make_path("check", end_slash=False),
                    query={"vault": vault, "permission": permission}
                )

            def grant(
                    self,
                    vault: str,
                    permission: int,
            ) -> JsonDict:
                """Grant this client a permission on a vault.

                Args:
                    vault: Vault label.
                    permission: Permission level to grant.

                Returns:
                    The resulting access record.
                """
                return self._post_json(self.make_path("grant", end_slash=False), json={
                    "vault": vault,
                    "permission": permission,
                })

            def revoke(
                    self,
                    vault: str,
                    permission: int,
            ) -> JsonDict:
                """Revoke this client's permission on a vault.

                Args:
                    vault: Vault label.
                    permission: Permission level to revoke.

                Returns:
                    The resulting access record.
                """
                return self._post_json(self.make_path("revoke", end_slash=False), json={
                    "vault": vault,
                    "permission": permission,
                })

            def update(
                    self,
                    vault: str,
                    permission: int,
            ) -> JsonDict:
                """Set this client's permission on a vault (upsert).

                PATCHes the access collection directly.

                Args:
                    vault: Vault label.
                    permission: Permission level to set.

                Returns:
                    The resulting access record.
                """
                return self._patch_json(self.make_path(end_slash=True), json={
                    "vault": vault,
                    "permission": permission,
                })

path = 'clients' class-attribute instance-attribute

Client

Bases: Resource

Campus Auth Client resource.

Source code in campus_python/auth/v1/clients.py

class Client(Resource):
    """Campus Auth Client resource."""

    @property
    def access(self) -> "Clients.Client.ClientAccess":
        return Clients.Client.ClientAccess("access", parent=self)

    @property
    def client_id(self) -> str:
        """The client_id of the resource was passed in as a path
        parameter, and is extracted using this property alias.
        """
        return self.path.strip("/")

    def delete(self) -> None:
        """Delete this client."""
        self._delete_json(self.make_path())

    def get(self) -> campus.model.Client:
        """Get this client.

        Returns:
            The Client.
        """
        body = self._get_json(self.make_path())
        return campus.model.Client.from_resource(body)

    def revoke(self) -> str:
        """Revoke the client secret and return the new one.

        Returns:
            The newly generated client secret.
        """
        return self._post_json(self.make_path("revoke", end_slash=False))["secret"]

    def update(
            self,
            name: str | None = None,
            description: str | None = None,
            redirect_uris: list[str] | None = None,
            allowed_scopes: list[str] | None = None,
            upstream_scopes: dict[str, list[str]] | None = None,
            token_bridge: bool | None = None
    ) -> campus.model.Client:
        """Update the client.

        Only the fields passed are sent; the server replaces each
        provided field wholesale (PATCH semantics), so list and
        dict values must carry the full desired contents.

        Args:
            name: New client name
            description: New client description
            redirect_uris: New OAuth redirect URIs
            allowed_scopes: Replacement token-scope allowlist
                (fail-closed: an empty list grants nothing)
            upstream_scopes: Replacement per-provider upstream
                scope map (e.g. {"google": [...]})
            token_bridge: Token bridge access flag (rejected by the
                server for public clients)

        Returns:
            The updated Client
        """
        return campus.model.Client.from_resource(self._patch_json(
            self.make_path(),
            json=payload(
                name=name,
                description=description,
                redirect_uris=redirect_uris,
                allowed_scopes=allowed_scopes,
                upstream_scopes=upstream_scopes,
                token_bridge=token_bridge,
            )
        ))

    class ClientAccess(Resource):
        """Campus Auth Client Access resource."""

        def get(
                self,
                vault: str | None = None
        ) -> JsonDict:
            """Get this client's vault access mapping.

            Args:
                vault: Scope the mapping to one vault label.

            Returns:
                {vault_label: permission} for the accessible vaults.
            """
            # GET rule is /<client_id>/access/ (trailing slash) and the
            # auth app sets strict_slashes — the slash is required.
            if vault:
                return self._get_json(
                    self.make_path(end_slash=True),
                    query={"vault": vault}
                )
            return self._get_json(self.make_path(end_slash=True))

        def check(
                self,
                vault: str,
                permission: int,
        ) -> JsonDict:
            """Check whether the client holds a permission on a vault.

            GET /<client_id>/access/check?vault=&permission=
            Returns: {"vault": ..., "permission": <bool>}
            """
            return self._get_json(
                self.make_path("check", end_slash=False),
                query={"vault": vault, "permission": permission}
            )

        def grant(
                self,
                vault: str,
                permission: int,
        ) -> JsonDict:
            """Grant this client a permission on a vault.

            Args:
                vault: Vault label.
                permission: Permission level to grant.

            Returns:
                The resulting access record.
            """
            return self._post_json(self.make_path("grant", end_slash=False), json={
                "vault": vault,
                "permission": permission,
            })

        def revoke(
                self,
                vault: str,
                permission: int,
        ) -> JsonDict:
            """Revoke this client's permission on a vault.

            Args:
                vault: Vault label.
                permission: Permission level to revoke.

            Returns:
                The resulting access record.
            """
            return self._post_json(self.make_path("revoke", end_slash=False), json={
                "vault": vault,
                "permission": permission,
            })

        def update(
                self,
                vault: str,
                permission: int,
        ) -> JsonDict:
            """Set this client's permission on a vault (upsert).

            PATCHes the access collection directly.

            Args:
                vault: Vault label.
                permission: Permission level to set.

            Returns:
                The resulting access record.
            """
            return self._patch_json(self.make_path(end_slash=True), json={
                "vault": vault,
                "permission": permission,
            })

access: 'Clients.Client.ClientAccess' property

client_id: str property

The client_id of the resource was passed in as a path parameter, and is extracted using this property alias.

ClientAccess

Bases: Resource

Campus Auth Client Access resource.

Source code in campus_python/auth/v1/clients.py

class ClientAccess(Resource):
    """Campus Auth Client Access resource."""

    def get(
            self,
            vault: str | None = None
    ) -> JsonDict:
        """Get this client's vault access mapping.

        Args:
            vault: Scope the mapping to one vault label.

        Returns:
            {vault_label: permission} for the accessible vaults.
        """
        # GET rule is /<client_id>/access/ (trailing slash) and the
        # auth app sets strict_slashes — the slash is required.
        if vault:
            return self._get_json(
                self.make_path(end_slash=True),
                query={"vault": vault}
            )
        return self._get_json(self.make_path(end_slash=True))

    def check(
            self,
            vault: str,
            permission: int,
    ) -> JsonDict:
        """Check whether the client holds a permission on a vault.

        GET /<client_id>/access/check?vault=&permission=
        Returns: {"vault": ..., "permission": <bool>}
        """
        return self._get_json(
            self.make_path("check", end_slash=False),
            query={"vault": vault, "permission": permission}
        )

    def grant(
            self,
            vault: str,
            permission: int,
    ) -> JsonDict:
        """Grant this client a permission on a vault.

        Args:
            vault: Vault label.
            permission: Permission level to grant.

        Returns:
            The resulting access record.
        """
        return self._post_json(self.make_path("grant", end_slash=False), json={
            "vault": vault,
            "permission": permission,
        })

    def revoke(
            self,
            vault: str,
            permission: int,
    ) -> JsonDict:
        """Revoke this client's permission on a vault.

        Args:
            vault: Vault label.
            permission: Permission level to revoke.

        Returns:
            The resulting access record.
        """
        return self._post_json(self.make_path("revoke", end_slash=False), json={
            "vault": vault,
            "permission": permission,
        })

    def update(
            self,
            vault: str,
            permission: int,
    ) -> JsonDict:
        """Set this client's permission on a vault (upsert).

        PATCHes the access collection directly.

        Args:
            vault: Vault label.
            permission: Permission level to set.

        Returns:
            The resulting access record.
        """
        return self._patch_json(self.make_path(end_slash=True), json={
            "vault": vault,
            "permission": permission,
        })

get(vault: str | None = None) -> JsonDict

Get this client's vault access mapping.

Parameters:

Name	Type	Description	Default
vault	str | None	Scope the mapping to one vault label.	None

Returns:

Type	Description
JsonDict	{vault_label: permission} for the accessible vaults.

Source code in campus_python/auth/v1/clients.py

def get(
        self,
        vault: str | None = None
) -> JsonDict:
    """Get this client's vault access mapping.

    Args:
        vault: Scope the mapping to one vault label.

    Returns:
        {vault_label: permission} for the accessible vaults.
    """
    # GET rule is /<client_id>/access/ (trailing slash) and the
    # auth app sets strict_slashes — the slash is required.
    if vault:
        return self._get_json(
            self.make_path(end_slash=True),
            query={"vault": vault}
        )
    return self._get_json(self.make_path(end_slash=True))

check(vault: str, permission: int) -> JsonDict

Check whether the client holds a permission on a vault.

GET //access/check?vault=&permission= Returns: {"vault": ..., "permission": }

Source code in campus_python/auth/v1/clients.py

def check(
        self,
        vault: str,
        permission: int,
) -> JsonDict:
    """Check whether the client holds a permission on a vault.

    GET /<client_id>/access/check?vault=&permission=
    Returns: {"vault": ..., "permission": <bool>}
    """
    return self._get_json(
        self.make_path("check", end_slash=False),
        query={"vault": vault, "permission": permission}
    )

grant(vault: str, permission: int) -> JsonDict

Grant this client a permission on a vault.

Parameters:

Name	Type	Description	Default
vault	str	Vault label.	required
permission	int	Permission level to grant.	required

Returns:

Type	Description
JsonDict	The resulting access record.

Source code in campus_python/auth/v1/clients.py

def grant(
        self,
        vault: str,
        permission: int,
) -> JsonDict:
    """Grant this client a permission on a vault.

    Args:
        vault: Vault label.
        permission: Permission level to grant.

    Returns:
        The resulting access record.
    """
    return self._post_json(self.make_path("grant", end_slash=False), json={
        "vault": vault,
        "permission": permission,
    })

revoke(vault: str, permission: int) -> JsonDict

Revoke this client's permission on a vault.

Parameters:

Name	Type	Description	Default
vault	str	Vault label.	required
permission	int	Permission level to revoke.	required

Returns:

Type	Description
JsonDict	The resulting access record.

Source code in campus_python/auth/v1/clients.py

def revoke(
        self,
        vault: str,
        permission: int,
) -> JsonDict:
    """Revoke this client's permission on a vault.

    Args:
        vault: Vault label.
        permission: Permission level to revoke.

    Returns:
        The resulting access record.
    """
    return self._post_json(self.make_path("revoke", end_slash=False), json={
        "vault": vault,
        "permission": permission,
    })

update(vault: str, permission: int) -> JsonDict

Set this client's permission on a vault (upsert).

PATCHes the access collection directly.

Parameters:

Name	Type	Description	Default
vault	str	Vault label.	required
permission	int	Permission level to set.	required

Returns:

Type	Description
JsonDict	The resulting access record.

Source code in campus_python/auth/v1/clients.py

def update(
        self,
        vault: str,
        permission: int,
) -> JsonDict:
    """Set this client's permission on a vault (upsert).

    PATCHes the access collection directly.

    Args:
        vault: Vault label.
        permission: Permission level to set.

    Returns:
        The resulting access record.
    """
    return self._patch_json(self.make_path(end_slash=True), json={
        "vault": vault,
        "permission": permission,
    })

delete() -> None

Delete this client.

Source code in campus_python/auth/v1/clients.py

def delete(self) -> None:
    """Delete this client."""
    self._delete_json(self.make_path())

get() -> campus.model.Client

Get this client.

Returns:

Type	Description
Client	The Client.

Source code in campus_python/auth/v1/clients.py

def get(self) -> campus.model.Client:
    """Get this client.

    Returns:
        The Client.
    """
    body = self._get_json(self.make_path())
    return campus.model.Client.from_resource(body)

revoke() -> str

Revoke the client secret and return the new one.

Returns:

Type	Description
str	The newly generated client secret.

Source code in campus_python/auth/v1/clients.py

def revoke(self) -> str:
    """Revoke the client secret and return the new one.

    Returns:
        The newly generated client secret.
    """
    return self._post_json(self.make_path("revoke", end_slash=False))["secret"]

update(name: str | None = None, description: str | None = None, redirect_uris: list[str] | None = None, allowed_scopes: list[str] | None = None, upstream_scopes: dict[str, list[str]] | None = None, token_bridge: bool | None = None) -> campus.model.Client

Update the client.

Only the fields passed are sent; the server replaces each provided field wholesale (PATCH semantics), so list and dict values must carry the full desired contents.

Parameters:

Name	Type	Description	Default
name	str | None	New client name	None
description	str | None	New client description	None
redirect_uris	list[str] | None	New OAuth redirect URIs	None
allowed_scopes	list[str] | None	Replacement token-scope allowlist (fail-closed: an empty list grants nothing)	None
upstream_scopes	dict[str, list[str]] | None	Replacement per-provider upstream scope map (e.g. {"google": [...]})	None
token_bridge	bool | None	Token bridge access flag (rejected by the server for public clients)	None

Returns:

Type	Description
Client	The updated Client

Source code in campus_python/auth/v1/clients.py

def update(
        self,
        name: str | None = None,
        description: str | None = None,
        redirect_uris: list[str] | None = None,
        allowed_scopes: list[str] | None = None,
        upstream_scopes: dict[str, list[str]] | None = None,
        token_bridge: bool | None = None
) -> campus.model.Client:
    """Update the client.

    Only the fields passed are sent; the server replaces each
    provided field wholesale (PATCH semantics), so list and
    dict values must carry the full desired contents.

    Args:
        name: New client name
        description: New client description
        redirect_uris: New OAuth redirect URIs
        allowed_scopes: Replacement token-scope allowlist
            (fail-closed: an empty list grants nothing)
        upstream_scopes: Replacement per-provider upstream
            scope map (e.g. {"google": [...]})
        token_bridge: Token bridge access flag (rejected by the
            server for public clients)

    Returns:
        The updated Client
    """
    return campus.model.Client.from_resource(self._patch_json(
        self.make_path(),
        json=payload(
            name=name,
            description=description,
            redirect_uris=redirect_uris,
            allowed_scopes=allowed_scopes,
            upstream_scopes=upstream_scopes,
            token_bridge=token_bridge,
        )
    ))

__getitem__(client_id: str) -> 'Clients.Client'

Get a specific client resource by ID.

Source code in campus_python/auth/v1/clients.py

def __getitem__(self, client_id: str) -> "Clients.Client":
    """Get a specific client resource by ID."""
    return Clients.Client(client_id, parent=self)

list() -> 'list[campus.model.Client]'

List all OAuth clients.

Returns:

Type	Description
'list[campus.model.Client]'	List of Client objects.

Source code in campus_python/auth/v1/clients.py

def list(self) -> "list[campus.model.Client]":
    """List all OAuth clients.

    Returns:
        List of Client objects.
    """
    body = self._get_json(self.make_path())
    return [
        campus.model.Client.from_resource(item)
        for item in body["clients"]
    ]

new(name: str, description: str, is_public: bool = False, redirect_uris: list[str] | None = None, allowed_scopes: list[str] | None = None, upstream_scopes: dict[str, list[str]] | None = None, token_bridge: bool | None = None) -> campus.model.Client

Create a new client.

Parameters:

Name	Type	Description	Default
name	str	Client name	required
description	str	Client description	required
is_public	bool	True for public clients (CLI/mobile apps) without secrets	False
redirect_uris	list[str] | None	OAuth redirect URIs for public clients	None
allowed_scopes	list[str] | None	Token-scope allowlist (fail-closed: an empty list grants nothing)	None
upstream_scopes	dict[str, list[str]] | None	Per-provider upstream scope map (e.g. {"google": [...]})	None
token_bridge	bool | None	Token bridge access flag (upstream token release; rejected by the server for public clients)	None

Returns:

Type	Description
Client	The created Client

Source code in campus_python/auth/v1/clients.py

def new(
        self,
        name: str,
        description: str,
        is_public: bool = False,
        redirect_uris: list[str] | None = None,
        allowed_scopes: list[str] | None = None,
        upstream_scopes: dict[str, list[str]] | None = None,
        token_bridge: bool | None = None
) -> campus.model.Client:
    """Create a new client.

    Args:
        name: Client name
        description: Client description
        is_public: True for public clients (CLI/mobile apps) without secrets
        redirect_uris: OAuth redirect URIs for public clients
        allowed_scopes: Token-scope allowlist (fail-closed: an
            empty list grants nothing)
        upstream_scopes: Per-provider upstream scope map
            (e.g. {"google": [...]})
        token_bridge: Token bridge access flag (upstream token
            release; rejected by the server for public clients)

    Returns:
        The created Client
    """
    return campus.model.Client.from_resource(self._post_json(
        self.make_path(),
        json=payload(
            name=name,
            description=description,
            is_public=is_public,
            redirect_uris=redirect_uris or [],
            allowed_scopes=allowed_scopes,
            upstream_scopes=upstream_scopes,
            token_bridge=token_bridge,
        )
    ))

Credentials

campus_python.auth.v1.credentials.Credentials

Bases: ResourceCollection

Campus Auth Credentials resource.

Usage: - creds = Credentials(root=...) - creds["github"].get(token_id) -> get by token id - creds["github"][user_id].get(client_id) -> get credentials for user

Source code in campus_python/auth/v1/credentials.py

class Credentials(ResourceCollection):
    """Campus Auth Credentials resource.

    Usage:
    - `creds = Credentials(root=...)`
    - `creds["github"].get(token_id)` -> get by token id
    - `creds["github"][user_id].get(client_id)` -> get credentials for user
    """
    path = "credentials"

    def __getitem__(self, provider: str) -> "Credentials.Provider":
        """Get provider-level collection."""
        return Credentials.Provider(provider, parent=self)

    class Provider(Resource):
        """Provider-level credentials resource.

        Instance `path` is set to include the provider name so that
        requests are made to /{base}/{credentials}/{provider}.
        """

        def __getitem__(self, user_id: str) -> "Credentials.Provider.User":
            """Get a user-specific credential resource."""
            return Credentials.Provider.User(user_id, parent=self)

        def get(
                self,
                token_id: str,
                include_token: bool = False,
        ) -> campus.model.UserCredentials:
            """Get one credential record.

            Reads are redacted by default (campus#1055): the joined
            token carries metadata only. include_token=True opts into
            the raw token — an operator-principal power on the
            service, honored on per-record reads only.
            """
            query: JsonDict = {"token_id": token_id}
            if include_token:
                query["include"] = "token"
            return campus.model.UserCredentials.from_resource(
                self._get_json(self.make_path(end_slash=True), query=query)
            )

        def list(self) -> "list[campus.model.UserCredentials]":
            """List credential records (admin view, campus#1055).

            Records carry token metadata only: the service never
            releases raw tokens in bulk, include opt-ins are honored
            on per-record reads exclusively.
            """
            body = self._get_json(self.make_path(end_slash=True))
            return [
                campus.model.UserCredentials.from_resource(item)
                for item in body.get("credentials", [])
            ]

        class User(Resource):
            """Credentials for a single user under a provider."""

            def delete(self, client_id: str | None = None) -> None:
                """Disconnect this user's credential record.

                Deletes the calling client's record by default;
                campus#1055 opens the body client_id to the operator
                so a deployment admin can disconnect an arbitrary
                integration's stored record for (provider, user).
                """
                target = client_id or env.CLIENT_ID
                self._delete_json(
                    self.make_path(end_slash=True),
                    json={"client_id": target}
                )

            def get(self, include_token: bool = False) -> campus.model.UserCredentials:
                """Get this user's credential record (redacted by
                default; include_token opts into the raw token,
                campus#1055)."""
                client_id = env.CLIENT_ID
                query: JsonDict = {"client_id": client_id}
                if include_token:
                    query["include"] = "token"
                return campus.model.UserCredentials.from_resource(
                    self._get_json(self.make_path(end_slash=True), query=query)
                )

            def new(
                    self,
                    scopes: "list[str]",
                    expires_in: int,
            ) -> campus.model.UserCredentials:
                """Issue new credentials for this provider and user.

                POST /credentials/<provider>/<user_id> with body
                {scopes, expires_in}; the client_id comes from the
                request's auth context, not the body.

                Returns:
                    The created UserCredentials
                """
                return campus.model.UserCredentials.from_resource(
                    self._post_json(
                        self.make_path(end_slash=True),
                        json={"scopes": scopes, "expires_in": expires_in}
                    )
                )

            def update(self, token: campus.model.OAuthToken) -> None:
                """Replace this user's stored token for the provider.

                The client_id rides on the body from the environment;
                the token is sent as its full resource form.

                Args:
                    token: The new OAuthToken to store.
                """
                client_id = env.CLIENT_ID
                json_data: JsonDict = {
                    "client_id": client_id,
                    "token": token.to_resource()
                }
                self._patch_json(self.make_path(end_slash=True), json=json_data)

path = 'credentials' class-attribute instance-attribute

Provider

Bases: Resource

Provider-level credentials resource.

Instance path is set to include the provider name so that requests are made to /{base}/{credentials}/{provider}.

Source code in campus_python/auth/v1/credentials.py

class Provider(Resource):
    """Provider-level credentials resource.

    Instance `path` is set to include the provider name so that
    requests are made to /{base}/{credentials}/{provider}.
    """

    def __getitem__(self, user_id: str) -> "Credentials.Provider.User":
        """Get a user-specific credential resource."""
        return Credentials.Provider.User(user_id, parent=self)

    def get(
            self,
            token_id: str,
            include_token: bool = False,
    ) -> campus.model.UserCredentials:
        """Get one credential record.

        Reads are redacted by default (campus#1055): the joined
        token carries metadata only. include_token=True opts into
        the raw token — an operator-principal power on the
        service, honored on per-record reads only.
        """
        query: JsonDict = {"token_id": token_id}
        if include_token:
            query["include"] = "token"
        return campus.model.UserCredentials.from_resource(
            self._get_json(self.make_path(end_slash=True), query=query)
        )

    def list(self) -> "list[campus.model.UserCredentials]":
        """List credential records (admin view, campus#1055).

        Records carry token metadata only: the service never
        releases raw tokens in bulk, include opt-ins are honored
        on per-record reads exclusively.
        """
        body = self._get_json(self.make_path(end_slash=True))
        return [
            campus.model.UserCredentials.from_resource(item)
            for item in body.get("credentials", [])
        ]

    class User(Resource):
        """Credentials for a single user under a provider."""

        def delete(self, client_id: str | None = None) -> None:
            """Disconnect this user's credential record.

            Deletes the calling client's record by default;
            campus#1055 opens the body client_id to the operator
            so a deployment admin can disconnect an arbitrary
            integration's stored record for (provider, user).
            """
            target = client_id or env.CLIENT_ID
            self._delete_json(
                self.make_path(end_slash=True),
                json={"client_id": target}
            )

        def get(self, include_token: bool = False) -> campus.model.UserCredentials:
            """Get this user's credential record (redacted by
            default; include_token opts into the raw token,
            campus#1055)."""
            client_id = env.CLIENT_ID
            query: JsonDict = {"client_id": client_id}
            if include_token:
                query["include"] = "token"
            return campus.model.UserCredentials.from_resource(
                self._get_json(self.make_path(end_slash=True), query=query)
            )

        def new(
                self,
                scopes: "list[str]",
                expires_in: int,
        ) -> campus.model.UserCredentials:
            """Issue new credentials for this provider and user.

            POST /credentials/<provider>/<user_id> with body
            {scopes, expires_in}; the client_id comes from the
            request's auth context, not the body.

            Returns:
                The created UserCredentials
            """
            return campus.model.UserCredentials.from_resource(
                self._post_json(
                    self.make_path(end_slash=True),
                    json={"scopes": scopes, "expires_in": expires_in}
                )
            )

        def update(self, token: campus.model.OAuthToken) -> None:
            """Replace this user's stored token for the provider.

            The client_id rides on the body from the environment;
            the token is sent as its full resource form.

            Args:
                token: The new OAuthToken to store.
            """
            client_id = env.CLIENT_ID
            json_data: JsonDict = {
                "client_id": client_id,
                "token": token.to_resource()
            }
            self._patch_json(self.make_path(end_slash=True), json=json_data)

User

Bases: Resource

Credentials for a single user under a provider.

Source code in campus_python/auth/v1/credentials.py

class User(Resource):
    """Credentials for a single user under a provider."""

    def delete(self, client_id: str | None = None) -> None:
        """Disconnect this user's credential record.

        Deletes the calling client's record by default;
        campus#1055 opens the body client_id to the operator
        so a deployment admin can disconnect an arbitrary
        integration's stored record for (provider, user).
        """
        target = client_id or env.CLIENT_ID
        self._delete_json(
            self.make_path(end_slash=True),
            json={"client_id": target}
        )

    def get(self, include_token: bool = False) -> campus.model.UserCredentials:
        """Get this user's credential record (redacted by
        default; include_token opts into the raw token,
        campus#1055)."""
        client_id = env.CLIENT_ID
        query: JsonDict = {"client_id": client_id}
        if include_token:
            query["include"] = "token"
        return campus.model.UserCredentials.from_resource(
            self._get_json(self.make_path(end_slash=True), query=query)
        )

    def new(
            self,
            scopes: "list[str]",
            expires_in: int,
    ) -> campus.model.UserCredentials:
        """Issue new credentials for this provider and user.

        POST /credentials/<provider>/<user_id> with body
        {scopes, expires_in}; the client_id comes from the
        request's auth context, not the body.

        Returns:
            The created UserCredentials
        """
        return campus.model.UserCredentials.from_resource(
            self._post_json(
                self.make_path(end_slash=True),
                json={"scopes": scopes, "expires_in": expires_in}
            )
        )

    def update(self, token: campus.model.OAuthToken) -> None:
        """Replace this user's stored token for the provider.

        The client_id rides on the body from the environment;
        the token is sent as its full resource form.

        Args:
            token: The new OAuthToken to store.
        """
        client_id = env.CLIENT_ID
        json_data: JsonDict = {
            "client_id": client_id,
            "token": token.to_resource()
        }
        self._patch_json(self.make_path(end_slash=True), json=json_data)

delete(client_id: str | None = None) -> None

Disconnect this user's credential record.

Deletes the calling client's record by default; campus#1055 opens the body client_id to the operator so a deployment admin can disconnect an arbitrary integration's stored record for (provider, user).

Source code in campus_python/auth/v1/credentials.py

def delete(self, client_id: str | None = None) -> None:
    """Disconnect this user's credential record.

    Deletes the calling client's record by default;
    campus#1055 opens the body client_id to the operator
    so a deployment admin can disconnect an arbitrary
    integration's stored record for (provider, user).
    """
    target = client_id or env.CLIENT_ID
    self._delete_json(
        self.make_path(end_slash=True),
        json={"client_id": target}
    )

get(include_token: bool = False) -> campus.model.UserCredentials

Get this user's credential record (redacted by default; include_token opts into the raw token, campus#1055).

Source code in campus_python/auth/v1/credentials.py

def get(self, include_token: bool = False) -> campus.model.UserCredentials:
    """Get this user's credential record (redacted by
    default; include_token opts into the raw token,
    campus#1055)."""
    client_id = env.CLIENT_ID
    query: JsonDict = {"client_id": client_id}
    if include_token:
        query["include"] = "token"
    return campus.model.UserCredentials.from_resource(
        self._get_json(self.make_path(end_slash=True), query=query)
    )

new(scopes: list[str], expires_in: int) -> campus.model.UserCredentials

Issue new credentials for this provider and user.

POST /credentials// with body {scopes, expires_in}; the client_id comes from the request's auth context, not the body.

Returns:

Type	Description
UserCredentials	The created UserCredentials

Source code in campus_python/auth/v1/credentials.py

def new(
        self,
        scopes: "list[str]",
        expires_in: int,
) -> campus.model.UserCredentials:
    """Issue new credentials for this provider and user.

    POST /credentials/<provider>/<user_id> with body
    {scopes, expires_in}; the client_id comes from the
    request's auth context, not the body.

    Returns:
        The created UserCredentials
    """
    return campus.model.UserCredentials.from_resource(
        self._post_json(
            self.make_path(end_slash=True),
            json={"scopes": scopes, "expires_in": expires_in}
        )
    )

update(token: campus.model.OAuthToken) -> None

Replace this user's stored token for the provider.

The client_id rides on the body from the environment; the token is sent as its full resource form.

Parameters:

Name	Type	Description	Default
token	OAuthToken	The new OAuthToken to store.	required

Source code in campus_python/auth/v1/credentials.py

def update(self, token: campus.model.OAuthToken) -> None:
    """Replace this user's stored token for the provider.

    The client_id rides on the body from the environment;
    the token is sent as its full resource form.

    Args:
        token: The new OAuthToken to store.
    """
    client_id = env.CLIENT_ID
    json_data: JsonDict = {
        "client_id": client_id,
        "token": token.to_resource()
    }
    self._patch_json(self.make_path(end_slash=True), json=json_data)

__getitem__(user_id: str) -> Credentials.Provider.User

Get a user-specific credential resource.

Source code in campus_python/auth/v1/credentials.py

def __getitem__(self, user_id: str) -> "Credentials.Provider.User":
    """Get a user-specific credential resource."""
    return Credentials.Provider.User(user_id, parent=self)

get(token_id: str, include_token: bool = False) -> campus.model.UserCredentials

Get one credential record.

Reads are redacted by default (campus#1055): the joined token carries metadata only. include_token=True opts into the raw token — an operator-principal power on the service, honored on per-record reads only.

Source code in campus_python/auth/v1/credentials.py

def get(
        self,
        token_id: str,
        include_token: bool = False,
) -> campus.model.UserCredentials:
    """Get one credential record.

    Reads are redacted by default (campus#1055): the joined
    token carries metadata only. include_token=True opts into
    the raw token — an operator-principal power on the
    service, honored on per-record reads only.
    """
    query: JsonDict = {"token_id": token_id}
    if include_token:
        query["include"] = "token"
    return campus.model.UserCredentials.from_resource(
        self._get_json(self.make_path(end_slash=True), query=query)
    )

list() -> list[campus.model.UserCredentials]

List credential records (admin view, campus#1055).

Records carry token metadata only: the service never releases raw tokens in bulk, include opt-ins are honored on per-record reads exclusively.

Source code in campus_python/auth/v1/credentials.py

def list(self) -> "list[campus.model.UserCredentials]":
    """List credential records (admin view, campus#1055).

    Records carry token metadata only: the service never
    releases raw tokens in bulk, include opt-ins are honored
    on per-record reads exclusively.
    """
    body = self._get_json(self.make_path(end_slash=True))
    return [
        campus.model.UserCredentials.from_resource(item)
        for item in body.get("credentials", [])
    ]

__getitem__(provider: str) -> Credentials.Provider

Get provider-level collection.

Source code in campus_python/auth/v1/credentials.py

def __getitem__(self, provider: str) -> "Credentials.Provider":
    """Get provider-level collection."""
    return Credentials.Provider(provider, parent=self)

Users

campus_python.auth.v1.users.Users

Bases: ResourceCollection

Vault Users resource.

Source code in campus_python/auth/v1/users.py

class Users(ResourceCollection):
    """Vault Users resource."""
    path = "users"

    def __getitem__(self, user_id: str) -> "Users.User":
        """Get a specific user resource by ID."""
        return Users.User(user_id, parent=self)

    def list(self) -> "list[campus.model.User]":
        """List all users.

        Returns:
            List of User objects.
        """
        body = self._get_json(self.make_path())
        return [
            campus.model.User.from_resource(item)
            for item in body["users"]
        ]

    def new(self, *, email: str, name: str) -> campus.model.User:
        """Create a user.

        A user's id IS its email.

        Args:
            email: The user's email (also their id).
            name: Display name.

        Returns:
            The created User.
        """
        return campus.model.User.from_resource(self._post_json(self.make_path(), json={
            "email": email,
            "name": name,
        }))

    class User(Resource):
        """Single vault user resource."""

        def activate(self) -> campus.model.User:
            """Activate this user (allows login).

            Returns:
                The activated User.
            """
            return campus.model.User.from_resource(
                self._post_json(self.make_path("activate", end_slash=False))
            )

        def delete(self) -> None:
            """Delete this user."""
            self._delete_json(self.make_path(end_slash=True))

        def get(self) -> campus.model.User:
            """Get this user.

            Returns:
                The User.
            """
            body = self._get_json(self.make_path(end_slash=True))
            return campus.model.User.from_resource(body)

        def update(self, *, name: str) -> campus.model.User:
            """Update the user's display name (campus-cli#42).

            The server PATCH accepts name only: a user's id IS its
            email (identity), and activation runs through the
            /activate endpoint, so no other field is patchable.

            Args:
                name: New display name

            Returns:
                The updated User
            """
            body = self._patch_json(
                self.make_path(end_slash=True), json={"name": name}
            )
            return campus.model.User.from_resource(body)

path = 'users' class-attribute instance-attribute

User

Bases: Resource

Single vault user resource.

Source code in campus_python/auth/v1/users.py

class User(Resource):
    """Single vault user resource."""

    def activate(self) -> campus.model.User:
        """Activate this user (allows login).

        Returns:
            The activated User.
        """
        return campus.model.User.from_resource(
            self._post_json(self.make_path("activate", end_slash=False))
        )

    def delete(self) -> None:
        """Delete this user."""
        self._delete_json(self.make_path(end_slash=True))

    def get(self) -> campus.model.User:
        """Get this user.

        Returns:
            The User.
        """
        body = self._get_json(self.make_path(end_slash=True))
        return campus.model.User.from_resource(body)

    def update(self, *, name: str) -> campus.model.User:
        """Update the user's display name (campus-cli#42).

        The server PATCH accepts name only: a user's id IS its
        email (identity), and activation runs through the
        /activate endpoint, so no other field is patchable.

        Args:
            name: New display name

        Returns:
            The updated User
        """
        body = self._patch_json(
            self.make_path(end_slash=True), json={"name": name}
        )
        return campus.model.User.from_resource(body)

activate() -> campus.model.User

Activate this user (allows login).

Returns:

Type	Description
User	The activated User.

Source code in campus_python/auth/v1/users.py

def activate(self) -> campus.model.User:
    """Activate this user (allows login).

    Returns:
        The activated User.
    """
    return campus.model.User.from_resource(
        self._post_json(self.make_path("activate", end_slash=False))
    )

delete() -> None

Delete this user.

Source code in campus_python/auth/v1/users.py

def delete(self) -> None:
    """Delete this user."""
    self._delete_json(self.make_path(end_slash=True))

get() -> campus.model.User

Get this user.

Returns:

Type	Description
User	The User.

Source code in campus_python/auth/v1/users.py

def get(self) -> campus.model.User:
    """Get this user.

    Returns:
        The User.
    """
    body = self._get_json(self.make_path(end_slash=True))
    return campus.model.User.from_resource(body)

update(*, name: str) -> campus.model.User

Update the user's display name (campus-cli#42).

The server PATCH accepts name only: a user's id IS its email (identity), and activation runs through the /activate endpoint, so no other field is patchable.

Parameters:

Name	Type	Description	Default
name	str	New display name	required

Returns:

Type	Description
User	The updated User

Source code in campus_python/auth/v1/users.py

def update(self, *, name: str) -> campus.model.User:
    """Update the user's display name (campus-cli#42).

    The server PATCH accepts name only: a user's id IS its
    email (identity), and activation runs through the
    /activate endpoint, so no other field is patchable.

    Args:
        name: New display name

    Returns:
        The updated User
    """
    body = self._patch_json(
        self.make_path(end_slash=True), json={"name": name}
    )
    return campus.model.User.from_resource(body)

__getitem__(user_id: str) -> Users.User

Get a specific user resource by ID.

Source code in campus_python/auth/v1/users.py

def __getitem__(self, user_id: str) -> "Users.User":
    """Get a specific user resource by ID."""
    return Users.User(user_id, parent=self)

list() -> list[campus.model.User]

List all users.

Returns:

Type	Description
list[User]	List of User objects.

Source code in campus_python/auth/v1/users.py

def list(self) -> "list[campus.model.User]":
    """List all users.

    Returns:
        List of User objects.
    """
    body = self._get_json(self.make_path())
    return [
        campus.model.User.from_resource(item)
        for item in body["users"]
    ]

new(*, email: str, name: str) -> campus.model.User

Create a user.

A user's id IS its email.

Parameters:

Name	Type	Description	Default
email	str	The user's email (also their id).	required
name	str	Display name.	required

Returns:

Type	Description
User	The created User.

Source code in campus_python/auth/v1/users.py

def new(self, *, email: str, name: str) -> campus.model.User:
    """Create a user.

    A user's id IS its email.

    Args:
        email: The user's email (also their id).
        name: Display name.

    Returns:
        The created User.
    """
    return campus.model.User.from_resource(self._post_json(self.make_path(), json={
        "email": email,
        "name": name,
    }))

Vaults

Vault values are addressed mapping-style: vaults["mylabel"]["key"], vaults["mylabel"]["key"] = value, del vaults["mylabel"]["key"]; missing keys raise KeyError.

campus_python.auth.v1.vaults.Vaults

Bases: ResourceCollection

Campus Auth Vaults resource.

Source code in campus_python/auth/v1/vaults.py

class Vaults(ResourceCollection):
    """Campus Auth Vaults resource."""
    path = "vaults"

    def __getitem__(self, label: str) -> "Vaults.Vault":
        return Vaults.Vault(label, parent=self)

    def labels(self) -> "list[str]":
        """List vault labels (admin view, campus#1055).

        GET /vaults/ — the operator, the super-admin root, or
        designated users holding any vault grant row plus the
        vaults:read scope; client principals hold per-label access
        only and cannot enumerate.
        """
        return self._get_json(self.make_path()).get("labels", [])

    class Vault(Resource):
        """A single vault (label) exposing mapping-like access."""

        def keys(self) -> "list[str]":
            """List the keys stored in this vault."""
            body = self._get_json(self.make_path(end_slash=True))
            return body["keys"]

        def __delitem__(self, key: str) -> None:
            resp = self.client.delete(self.make_path(key, end_slash=True))
            if resp.status_code == 404:
                raise KeyError(key)
            resp.raise_for_status()
            return None

        def __getitem__(self, key: str) -> str:
            resp = self.client.get(self.make_path(key, end_slash=True))
            # translate 404 into KeyError for route compatibility
            if resp.status_code == 404:
                raise KeyError(key)
            resp.raise_for_status()
            body = resp.json()
            # expected shape: {"key": value}
            return body["key"]

        def __setitem__(self, key: str, value: Any) -> None:
            self._post_json(self.make_path(key, end_slash=True), json={"value": value})
            return None

path = 'vaults' class-attribute instance-attribute

Vault

Bases: Resource

A single vault (label) exposing mapping-like access.

Source code in campus_python/auth/v1/vaults.py

class Vault(Resource):
    """A single vault (label) exposing mapping-like access."""

    def keys(self) -> "list[str]":
        """List the keys stored in this vault."""
        body = self._get_json(self.make_path(end_slash=True))
        return body["keys"]

    def __delitem__(self, key: str) -> None:
        resp = self.client.delete(self.make_path(key, end_slash=True))
        if resp.status_code == 404:
            raise KeyError(key)
        resp.raise_for_status()
        return None

    def __getitem__(self, key: str) -> str:
        resp = self.client.get(self.make_path(key, end_slash=True))
        # translate 404 into KeyError for route compatibility
        if resp.status_code == 404:
            raise KeyError(key)
        resp.raise_for_status()
        body = resp.json()
        # expected shape: {"key": value}
        return body["key"]

    def __setitem__(self, key: str, value: Any) -> None:
        self._post_json(self.make_path(key, end_slash=True), json={"value": value})
        return None

keys() -> list[str]

List the keys stored in this vault.

Source code in campus_python/auth/v1/vaults.py

def keys(self) -> "list[str]":
    """List the keys stored in this vault."""
    body = self._get_json(self.make_path(end_slash=True))
    return body["keys"]

__delitem__(key: str) -> None

Source code in campus_python/auth/v1/vaults.py

def __delitem__(self, key: str) -> None:
    resp = self.client.delete(self.make_path(key, end_slash=True))
    if resp.status_code == 404:
        raise KeyError(key)
    resp.raise_for_status()
    return None

__getitem__(key: str) -> str

Source code in campus_python/auth/v1/vaults.py

def __getitem__(self, key: str) -> str:
    resp = self.client.get(self.make_path(key, end_slash=True))
    # translate 404 into KeyError for route compatibility
    if resp.status_code == 404:
        raise KeyError(key)
    resp.raise_for_status()
    body = resp.json()
    # expected shape: {"key": value}
    return body["key"]

__setitem__(key: str, value: Any) -> None

Source code in campus_python/auth/v1/vaults.py

def __setitem__(self, key: str, value: Any) -> None:
    self._post_json(self.make_path(key, end_slash=True), json={"value": value})
    return None

__getitem__(label: str) -> Vaults.Vault

Source code in campus_python/auth/v1/vaults.py

def __getitem__(self, label: str) -> "Vaults.Vault":
    return Vaults.Vault(label, parent=self)

labels() -> list[str]

List vault labels (admin view, campus#1055).

GET /vaults/ — the operator, the super-admin root, or designated users holding any vault grant row plus the vaults:read scope; client principals hold per-label access only and cannot enumerate.

Source code in campus_python/auth/v1/vaults.py

def labels(self) -> "list[str]":
    """List vault labels (admin view, campus#1055).

    GET /vaults/ — the operator, the super-admin root, or
    designated users holding any vault grant row plus the
    vaults:read scope; client principals hold per-label access
    only and cannot enumerate.
    """
    return self._get_json(self.make_path()).get("labels", [])

Grants

The access-grant store: who can administer which vocabulary, and vault bitflag access.

campus_python.auth.v1.grants.Grants

Bases: ResourceCollection

Campus Auth access-grant store.

Source code in campus_python/auth/v1/grants.py

class Grants(ResourceCollection):
    """Campus Auth access-grant store."""
    path = "grants"

    def list(
            self,
            *,
            grantee_type: "str | None" = None,
            grantee_id: "str | None" = None,
            resource_type: "str | None" = None,
            resource_id: "str | None" = None,
    ) -> JsonDict:
        """List access-grant rows.

        Authorization: unfiltered lists require the operator; user
        principals may list only the single vocabulary they
        administer.

        Args:
            grantee_type: Filter by "user" or "client"
            grantee_id: Filter by grantee identity (user email / client id)
            resource_type: Filter by vocabulary ("users", "clients",
                "vault", ...)
            resource_id: Filter by instance (vault label)

        Returns:
            {"grants": [...]}
        """
        query: JsonDict = {}
        if grantee_type is not None:
            query["grantee_type"] = grantee_type
        if grantee_id is not None:
            query["grantee_id"] = grantee_id
        if resource_type is not None:
            query["resource_type"] = resource_type
        if resource_id is not None:
            query["resource_id"] = resource_id
        return self._get_json(self.make_path(), query=query or None)

    def check(
            self,
            *,
            grantee_type: str,
            grantee_id: str,
            resource_type: str,
            resource_id: str = "",
            bits: "int | None" = None,
            level: "str | None" = None,
    ) -> bool:
        """Check whether a grantee's row covers a permission.

        Authorization: operator, or a same-vocabulary admin.

        Args:
            grantee_type: "user" or "client"
            grantee_id: Grantee identity
            resource_type: Vocabulary ("users", "clients", "vault", ...)
            resource_id: Instance (vault label; default none)
            bits: Vault bitflag mask to test
            level: Management level to test

        Returns:
            True when the held row covers the requested permission
        """
        query: JsonDict = {
            "grantee_type": grantee_type,
            "grantee_id": grantee_id,
            "resource_type": resource_type,
        }
        if resource_id:
            query["resource_id"] = resource_id
        if bits is not None:
            query["bits"] = bits
        if level is not None:
            query["level"] = level
        body = self._get_json(self.make_path("check", end_slash=False), query=query)
        return bool(body.get("granted"))

    def grant(
            self,
            *,
            grantee_type: str,
            grantee_id: str,
            resource_type: str,
            resource_id: str = "",
            bits: "int | None" = None,
            level: "str | None" = None,
    ) -> JsonDict:
        """Grant access (create a row, or raise an existing one).

        Authorization: operator, or a same-vocabulary admin. Vault
        bits OR into the existing mask; management levels never
        downgrade — revoke first to lower a level.

        Anti-escalation (#883): no principal may administer its own
        grants, and vocabulary-level clients:admin rows are rejected
        (operator-equivalent).

        Args:
            grantee_type: "user" or "client"
            grantee_id: Grantee identity
            resource_type: Vocabulary ("users", "clients", "vault", ...)
            resource_id: Instance (required for vault labels)
            bits: Vault bitflag mask (vaults only)
            level: Management level (read/mod/write/admin)

        Returns:
            The grant row (post-application state)
        """
        # resource_id or None: "" means "no instance" here, and
        # payload only drops None.
        return self._post_json(self.make_path(), json=payload(
            grantee_type=grantee_type,
            grantee_id=grantee_id,
            resource_type=resource_type,
            resource_id=resource_id or None,
            bits=bits,
            level=level,
        ))["grant"]

    def revoke(
            self,
            *,
            grantee_type: str,
            grantee_id: str,
            resource_type: str,
            resource_id: str = "",
            bits: "int | None" = None,
            level: "str | None" = None,
    ) -> "JsonDict | None":
        """Revoke access (clear vault bits, delete covered levels).

        Authorization: operator, or a same-vocabulary admin. Vault
        rows delete at zero bits; a revoked level's row is deleted
        when the held level covers it.

        Args:
            grantee_type: "user" or "client"
            grantee_id: Grantee identity
            resource_type: Vocabulary
            resource_id: Instance (vault label)
            bits: Vault bitflag mask to clear
            level: Management level to revoke

        Returns:
            The remaining grant row, or None when the row was deleted
        """
        return self._post_json(
            self.make_path("revoke", end_slash=False),
            json=payload(
                grantee_type=grantee_type,
                grantee_id=grantee_id,
                resource_type=resource_type,
                resource_id=resource_id or None,
                bits=bits,
                level=level,
            ),
        )["grant"]

    def delete(self, grant_id: str) -> None:
        """Delete a grant row outright by its id.

        Authorization: operator, or a same-vocabulary admin for the
        row's vocabulary.

        Args:
            grant_id: The grant row's id (from list())
        """
        self._delete_json(self.make_path(grant_id, end_slash=False))

path = 'grants' class-attribute instance-attribute

list(*, grantee_type: str | None = None, grantee_id: str | None = None, resource_type: str | None = None, resource_id: str | None = None) -> JsonDict

List access-grant rows.

Authorization: unfiltered lists require the operator; user principals may list only the single vocabulary they administer.

Parameters:

Name	Type	Description	Default
grantee_type	str | None	Filter by "user" or "client"	None
grantee_id	str | None	Filter by grantee identity (user email / client id)	None
resource_type	str | None	Filter by vocabulary ("users", "clients", "vault", ...)	None
resource_id	str | None	Filter by instance (vault label)	None

Returns:

Type	Description
JsonDict	{"grants": [...]}

Source code in campus_python/auth/v1/grants.py

def list(
        self,
        *,
        grantee_type: "str | None" = None,
        grantee_id: "str | None" = None,
        resource_type: "str | None" = None,
        resource_id: "str | None" = None,
) -> JsonDict:
    """List access-grant rows.

    Authorization: unfiltered lists require the operator; user
    principals may list only the single vocabulary they
    administer.

    Args:
        grantee_type: Filter by "user" or "client"
        grantee_id: Filter by grantee identity (user email / client id)
        resource_type: Filter by vocabulary ("users", "clients",
            "vault", ...)
        resource_id: Filter by instance (vault label)

    Returns:
        {"grants": [...]}
    """
    query: JsonDict = {}
    if grantee_type is not None:
        query["grantee_type"] = grantee_type
    if grantee_id is not None:
        query["grantee_id"] = grantee_id
    if resource_type is not None:
        query["resource_type"] = resource_type
    if resource_id is not None:
        query["resource_id"] = resource_id
    return self._get_json(self.make_path(), query=query or None)

check(*, grantee_type: str, grantee_id: str, resource_type: str, resource_id: str = '', bits: int | None = None, level: str | None = None) -> bool

Check whether a grantee's row covers a permission.

Authorization: operator, or a same-vocabulary admin.

Parameters:

Name	Type	Description	Default
grantee_type	str	"user" or "client"	required
grantee_id	str	Grantee identity	required
resource_type	str	Vocabulary ("users", "clients", "vault", ...)	required
resource_id	str	Instance (vault label; default none)	''
bits	int | None	Vault bitflag mask to test	None
level	str | None	Management level to test	None

Returns:

Type	Description
bool	True when the held row covers the requested permission

Source code in campus_python/auth/v1/grants.py

def check(
        self,
        *,
        grantee_type: str,
        grantee_id: str,
        resource_type: str,
        resource_id: str = "",
        bits: "int | None" = None,
        level: "str | None" = None,
) -> bool:
    """Check whether a grantee's row covers a permission.

    Authorization: operator, or a same-vocabulary admin.

    Args:
        grantee_type: "user" or "client"
        grantee_id: Grantee identity
        resource_type: Vocabulary ("users", "clients", "vault", ...)
        resource_id: Instance (vault label; default none)
        bits: Vault bitflag mask to test
        level: Management level to test

    Returns:
        True when the held row covers the requested permission
    """
    query: JsonDict = {
        "grantee_type": grantee_type,
        "grantee_id": grantee_id,
        "resource_type": resource_type,
    }
    if resource_id:
        query["resource_id"] = resource_id
    if bits is not None:
        query["bits"] = bits
    if level is not None:
        query["level"] = level
    body = self._get_json(self.make_path("check", end_slash=False), query=query)
    return bool(body.get("granted"))

grant(*, grantee_type: str, grantee_id: str, resource_type: str, resource_id: str = '', bits: int | None = None, level: str | None = None) -> JsonDict

Grant access (create a row, or raise an existing one).

Authorization: operator, or a same-vocabulary admin. Vault bits OR into the existing mask; management levels never downgrade — revoke first to lower a level.

Anti-escalation (#883): no principal may administer its own grants, and vocabulary-level clients:admin rows are rejected (operator-equivalent).

Parameters:

Name	Type	Description	Default
grantee_type	str	"user" or "client"	required
grantee_id	str	Grantee identity	required
resource_type	str	Vocabulary ("users", "clients", "vault", ...)	required
resource_id	str	Instance (required for vault labels)	''
bits	int | None	Vault bitflag mask (vaults only)	None
level	str | None	Management level (read/mod/write/admin)	None

Returns:

Type	Description
JsonDict	The grant row (post-application state)

Source code in campus_python/auth/v1/grants.py

def grant(
        self,
        *,
        grantee_type: str,
        grantee_id: str,
        resource_type: str,
        resource_id: str = "",
        bits: "int | None" = None,
        level: "str | None" = None,
) -> JsonDict:
    """Grant access (create a row, or raise an existing one).

    Authorization: operator, or a same-vocabulary admin. Vault
    bits OR into the existing mask; management levels never
    downgrade — revoke first to lower a level.

    Anti-escalation (#883): no principal may administer its own
    grants, and vocabulary-level clients:admin rows are rejected
    (operator-equivalent).

    Args:
        grantee_type: "user" or "client"
        grantee_id: Grantee identity
        resource_type: Vocabulary ("users", "clients", "vault", ...)
        resource_id: Instance (required for vault labels)
        bits: Vault bitflag mask (vaults only)
        level: Management level (read/mod/write/admin)

    Returns:
        The grant row (post-application state)
    """
    # resource_id or None: "" means "no instance" here, and
    # payload only drops None.
    return self._post_json(self.make_path(), json=payload(
        grantee_type=grantee_type,
        grantee_id=grantee_id,
        resource_type=resource_type,
        resource_id=resource_id or None,
        bits=bits,
        level=level,
    ))["grant"]

revoke(*, grantee_type: str, grantee_id: str, resource_type: str, resource_id: str = '', bits: int | None = None, level: str | None = None) -> JsonDict | None

Revoke access (clear vault bits, delete covered levels).

Authorization: operator, or a same-vocabulary admin. Vault rows delete at zero bits; a revoked level's row is deleted when the held level covers it.

Parameters:

Name	Type	Description	Default
grantee_type	str	"user" or "client"	required
grantee_id	str	Grantee identity	required
resource_type	str	Vocabulary	required
resource_id	str	Instance (vault label)	''
bits	int | None	Vault bitflag mask to clear	None
level	str | None	Management level to revoke	None

Returns:

Type	Description
JsonDict | None	The remaining grant row, or None when the row was deleted

Source code in campus_python/auth/v1/grants.py

def revoke(
        self,
        *,
        grantee_type: str,
        grantee_id: str,
        resource_type: str,
        resource_id: str = "",
        bits: "int | None" = None,
        level: "str | None" = None,
) -> "JsonDict | None":
    """Revoke access (clear vault bits, delete covered levels).

    Authorization: operator, or a same-vocabulary admin. Vault
    rows delete at zero bits; a revoked level's row is deleted
    when the held level covers it.

    Args:
        grantee_type: "user" or "client"
        grantee_id: Grantee identity
        resource_type: Vocabulary
        resource_id: Instance (vault label)
        bits: Vault bitflag mask to clear
        level: Management level to revoke

    Returns:
        The remaining grant row, or None when the row was deleted
    """
    return self._post_json(
        self.make_path("revoke", end_slash=False),
        json=payload(
            grantee_type=grantee_type,
            grantee_id=grantee_id,
            resource_type=resource_type,
            resource_id=resource_id or None,
            bits=bits,
            level=level,
        ),
    )["grant"]

delete(grant_id: str) -> None

Delete a grant row outright by its id.

Authorization: operator, or a same-vocabulary admin for the row's vocabulary.

Parameters:

Name	Type	Description	Default
grant_id	str	The grant row's id (from list())	required

Source code in campus_python/auth/v1/grants.py

def delete(self, grant_id: str) -> None:
    """Delete a grant row outright by its id.

    Authorization: operator, or a same-vocabulary admin for the
    row's vocabulary.

    Args:
        grant_id: The grant row's id (from list())
    """
    self._delete_json(self.make_path(grant_id, end_slash=False))

Connections

The user-facing view of upstream provider connections (metadata only — token values are broker-only).

campus_python.auth.v1.connections.Connections

Bases: ResourceCollection

Campus Auth Connections resource.

Source code in campus_python/auth/v1/connections.py

class Connections(ResourceCollection):
    """Campus Auth Connections resource."""
    path = "connections"

    def __getitem__(self, provider: str) -> "Connections.Connection":
        """Get a specific provider's connection resource."""
        return Connections.Connection(provider, parent=self)

    def list(
            self,
            *,
            user_id: "str | None" = None,
    ) -> "list[JsonDict]":
        """List the target user's upstream connections.

        Each entry carries {provider, integration, scopes, connected_at,
        expires_at} — never token values. Auth shape mirrors the server:
        a bearer token acts for its own user (user_id ignored), while
        basic (client-credentials) auth must name a user via user_id.

        Args:
            user_id: Delegated target user (required for basic auth)

        Returns:
            List of connection metadata dicts
        """
        query = {"user_id": user_id} if user_id else None
        return self._get_json(self.make_path(), query=query)["connections"]

    class Connection(Resource):
        """Single provider connection resource."""

        def __getitem__(
                self,
                integration: str,
        ) -> "Connections.Connection.Integration":
            """Get an integration-namespaced connection resource."""
            return Connections.Connection.Integration(
                integration, parent=self
            )

        def delete(self, *, user_id: "str | None" = None) -> None:
            """Disconnect every credential for this base provider.

            404 (NotFoundError) means there was nothing to disconnect;
            callers treating disconnect as idempotent can catch it.

            Args:
                user_id: Delegated target user (required for basic auth)
            """
            query = {"user_id": user_id} if user_id else None
            self._delete_json(self.make_path(end_slash=True), query=query)
            return None

        class Integration(Resource):
            """Integration-namespaced connection resource."""

            def delete(self, *, user_id: "str | None" = None) -> None:
                """Disconnect this provider integration (e.g. google/classroom).

                Args:
                    user_id: Delegated target user (required for basic auth)
                """
                query = {"user_id": user_id} if user_id else None
                self._delete_json(self.make_path(end_slash=True), query=query)
                return None

path = 'connections' class-attribute instance-attribute

Connection

Bases: Resource

Single provider connection resource.

Source code in campus_python/auth/v1/connections.py

class Connection(Resource):
    """Single provider connection resource."""

    def __getitem__(
            self,
            integration: str,
    ) -> "Connections.Connection.Integration":
        """Get an integration-namespaced connection resource."""
        return Connections.Connection.Integration(
            integration, parent=self
        )

    def delete(self, *, user_id: "str | None" = None) -> None:
        """Disconnect every credential for this base provider.

        404 (NotFoundError) means there was nothing to disconnect;
        callers treating disconnect as idempotent can catch it.

        Args:
            user_id: Delegated target user (required for basic auth)
        """
        query = {"user_id": user_id} if user_id else None
        self._delete_json(self.make_path(end_slash=True), query=query)
        return None

    class Integration(Resource):
        """Integration-namespaced connection resource."""

        def delete(self, *, user_id: "str | None" = None) -> None:
            """Disconnect this provider integration (e.g. google/classroom).

            Args:
                user_id: Delegated target user (required for basic auth)
            """
            query = {"user_id": user_id} if user_id else None
            self._delete_json(self.make_path(end_slash=True), query=query)
            return None

Integration

Bases: Resource

Integration-namespaced connection resource.

Source code in campus_python/auth/v1/connections.py

class Integration(Resource):
    """Integration-namespaced connection resource."""

    def delete(self, *, user_id: "str | None" = None) -> None:
        """Disconnect this provider integration (e.g. google/classroom).

        Args:
            user_id: Delegated target user (required for basic auth)
        """
        query = {"user_id": user_id} if user_id else None
        self._delete_json(self.make_path(end_slash=True), query=query)
        return None

delete(*, user_id: str | None = None) -> None

Disconnect this provider integration (e.g. google/classroom).

Parameters:

Name	Type	Description	Default
user_id	str | None	Delegated target user (required for basic auth)	None

Source code in campus_python/auth/v1/connections.py

def delete(self, *, user_id: "str | None" = None) -> None:
    """Disconnect this provider integration (e.g. google/classroom).

    Args:
        user_id: Delegated target user (required for basic auth)
    """
    query = {"user_id": user_id} if user_id else None
    self._delete_json(self.make_path(end_slash=True), query=query)
    return None

__getitem__(integration: str) -> Connections.Connection.Integration

Get an integration-namespaced connection resource.

Source code in campus_python/auth/v1/connections.py

def __getitem__(
        self,
        integration: str,
) -> "Connections.Connection.Integration":
    """Get an integration-namespaced connection resource."""
    return Connections.Connection.Integration(
        integration, parent=self
    )

delete(*, user_id: str | None = None) -> None

Disconnect every credential for this base provider.

404 (NotFoundError) means there was nothing to disconnect; callers treating disconnect as idempotent can catch it.

Parameters:

Name	Type	Description	Default
user_id	str | None	Delegated target user (required for basic auth)	None

Source code in campus_python/auth/v1/connections.py

def delete(self, *, user_id: "str | None" = None) -> None:
    """Disconnect every credential for this base provider.

    404 (NotFoundError) means there was nothing to disconnect;
    callers treating disconnect as idempotent can catch it.

    Args:
        user_id: Delegated target user (required for basic auth)
    """
    query = {"user_id": user_id} if user_id else None
    self._delete_json(self.make_path(end_slash=True), query=query)
    return None

__getitem__(provider: str) -> Connections.Connection

Get a specific provider's connection resource.

Source code in campus_python/auth/v1/connections.py

def __getitem__(self, provider: str) -> "Connections.Connection":
    """Get a specific provider's connection resource."""
    return Connections.Connection(provider, parent=self)

list(*, user_id: str | None = None) -> list[JsonDict]

List the target user's upstream connections.

Each entry carries {provider, integration, scopes, connected_at, expires_at} — never token values. Auth shape mirrors the server: a bearer token acts for its own user (user_id ignored), while basic (client-credentials) auth must name a user via user_id.

Parameters:

Name	Type	Description	Default
user_id	str | None	Delegated target user (required for basic auth)	None

Returns:

Type	Description
list[JsonDict]	List of connection metadata dicts

Source code in campus_python/auth/v1/connections.py

def list(
        self,
        *,
        user_id: "str | None" = None,
) -> "list[JsonDict]":
    """List the target user's upstream connections.

    Each entry carries {provider, integration, scopes, connected_at,
    expires_at} — never token values. Auth shape mirrors the server:
    a bearer token acts for its own user (user_id ignored), while
    basic (client-credentials) auth must name a user via user_id.

    Args:
        user_id: Delegated target user (required for basic auth)

    Returns:
        List of connection metadata dicts
    """
    query = {"user_id": user_id} if user_id else None
    return self._get_json(self.make_path(), query=query)["connections"]

Broker

The sanctioned release path for upstream access tokens (token_bridge clients).

campus_python.auth.v1.broker.Broker

Bases: ResourceRoot

Campus Auth token broker resource.

Source code in campus_python/auth/v1/broker.py

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}/")
        return self._post_json(path, json=payload(min_scopes=min_scopes))

url_prefix = '/auth/v1/broker' class-attribute instance-attribute

__init__(root: ResourceRoot)

Source code in campus_python/auth/v1/broker.py

def __init__(self, root: ResourceRoot):
    super().__init__(json_client=root.client)
    self._root = root

token(provider: str, integration: str | None = None, *, min_scopes: list[str] | None = None) -> JsonDict

Release the bearer's upstream access token for a provider.

Parameters:

Name	Type	Description	Default
provider	str	Base provider (e.g. "google")	required
integration	str | None	Integration slug for the namespaced route (e.g. "classroom" → /broker/google/classroom/)	None
min_scopes	list[str] | None	Scopes the caller requires; denied unless the user's upstream grant covers them and the client's upstream_scopes allowlist permits them	None

Returns:

Type	Description
JsonDict	{provider, user_id, access_token, token_type, expires_in,
JsonDict	scope} — the refresh token never leaves Campus

Raises:

Type	Description
NotFoundError	No connection for this provider (404); namespaced providers pointed at the identity route are redirected to the integration route the same way
AuthenticationError	Bearer session expired (401)
APIError	403 with details.missing_scopes, or 400 AUTH_INVALID_SCOPE / configuration problems

Source code in campus_python/auth/v1/broker.py

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}/")
    return self._post_json(path, json=payload(min_scopes=min_scopes))

Root

Operator-facing root actions (token/client introspection).

campus_python.auth.v1.root.Root

Bases: ResourceCollection

Campus Auth root resource.

Source code in campus_python/auth/v1/root.py

class Root(ResourceCollection):
    """Campus Auth root resource."""
    path = "root"

    def authenticate(
            self,
            *,
            token: str | None = None,
            client_id: schema.CampusID | None = None,
            client_secret: str | None = None,
    ) -> dict[str, Any]:
        """Authenticate the given token or client credentials.

        Returns: {
            "client": { ... }
        } or {
            "client": { ... },
            "user": { ... }
        }
        """
        if token:
            body = self._post_json(self.make_path(), json={"token": token})
        elif client_id and client_secret:
            body = self._post_json(
                self.make_path(),
                json={"client_id": client_id, "client_secret": client_secret}
            )
        else:
            raise errors.AuthenticationError(
                error_description="Missing or incomplete authentication information."
            )
        body["client"] = campus.model.Client.from_resource(body["client"])
        if "user" in body:
            body["user"] = campus.model.User.from_resource(body["user"])
        return body

path = 'root' class-attribute instance-attribute

authenticate(*, token: str | None = None, client_id: schema.CampusID | None = None, client_secret: str | None = None) -> dict[str, Any]

Authenticate the given token or client credentials.

{

Type	Description
dict[str, Any]	"client": { ... }

} or { "client": { ... }, "user": { ... } }

Source code in campus_python/auth/v1/root.py

def authenticate(
        self,
        *,
        token: str | None = None,
        client_id: schema.CampusID | None = None,
        client_secret: str | None = None,
) -> dict[str, Any]:
    """Authenticate the given token or client credentials.

    Returns: {
        "client": { ... }
    } or {
        "client": { ... },
        "user": { ... }
    }
    """
    if token:
        body = self._post_json(self.make_path(), json={"token": token})
    elif client_id and client_secret:
        body = self._post_json(
            self.make_path(),
            json={"client_id": client_id, "client_secret": client_secret}
        )
    else:
        raise errors.AuthenticationError(
            error_description="Missing or incomplete authentication information."
        )
    body["client"] = campus.model.Client.from_resource(body["client"])
    if "user" in body:
        body["user"] = campus.model.User.from_resource(body["user"])
    return body

Session-id seam

The session-id resolver flask_campus installs; ids resolve explicit argument > provider.

campus_python.auth.v1.sessionstore

campus.python.auth.v1.sessionstore

Client-side session-id resolution seam for the auth session/login lanes.

The SDK never touches a web-framework session store: hosts that keep their session ids in their own request state — e.g. flask_campus, which stores them under sessions.SESSION_STORAGE_KEY / logins.LOGIN_STORAGE_KEY — install a provider that resolves the current id per lane:

from campus_python.auth import v1
v1.set_session_id_provider(my_resolver)

The provider takes a lane name — "session" for auth sessions or "login" for login sessions — and returns the current session id or None. Every read resolves explicit argument first, then the provider; with neither, the resource methods raise AuthenticationError naming the seam to install.

set_session_id_provider(provider: typing.Callable[[str], str | None] | None) -> None

Install (or clear) the session-id resolver for both lanes.

The provider is consulted whenever a resource method needs the current session id and none was passed explicitly. Pass None to clear it (tests).

Source code in campus_python/auth/v1/sessionstore.py

def set_session_id_provider(
        provider: typing.Callable[[str], str | None] | None,
) -> None:
    """Install (or clear) the session-id resolver for both lanes.

    The provider is consulted whenever a resource method needs the
    current session id and none was passed explicitly. Pass None to
    clear it (tests).
    """
    global _session_id_provider
    _session_id_provider = provider