From ce77a7dec76423832ae69e70645f67f934258975 Mon Sep 17 00:00:00 2001
From: JS <kureshii@js-xiaomi.localdomain>
Date: Sat, 10 Oct 2026 23:27:17 +0800
Subject: [PATCH] feat(auth): Grants resource for the access-grant store

Wraps the /auth/v1/grants/ administration surface (#883/#886): list
(the who-can-administer-what matrix, with filters as query params),
check, grant, revoke, and delete-by-id. Rows are plain dicts (the
server's AccessGrant resource; no campus model exists for it).
campus-cli#50's grant commands build on this.
---
 campus_python/auth/v1/__init__.py |   9 ++
 campus_python/auth/v1/grants.py   | 200 ++++++++++++++++++++++++++++++
 tests/unit/test_auth_grants.py    | 174 ++++++++++++++++++++++++++
 3 files changed, 383 insertions(+)
 create mode 100644 campus_python/auth/v1/grants.py
 create mode 100644 tests/unit/test_auth_grants.py

diff --git a/campus_python/auth/v1/__init__.py b/campus_python/auth/v1/__init__.py
index 03384a7..c8a3cae 100644
--- a/campus_python/auth/v1/__init__.py
+++ b/campus_python/auth/v1/__init__.py
@@ -22,6 +22,7 @@
     clients,
     connections,
     credentials,
+    grants,
     logins,
     oauth,
     root,
@@ -49,6 +50,7 @@ def __init__(self, json_client: JsonClient):
         self._clients = None
         self._connections = None
         self._credentials = None
+        self._grants = None
         self._logins = None
         self._oauth = None
         self._root = None
@@ -127,6 +129,13 @@ def users(self) -> users.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."""
diff --git a/campus_python/auth/v1/grants.py b/campus_python/auth/v1/grants.py
new file mode 100644
index 0000000..4479ee5
--- /dev/null
+++ b/campus_python/auth/v1/grants.py
@@ -0,0 +1,200 @@
+"""campus.python.auth.v1.grants
+
+Campus Auth grants resource (v1).
+
+The access-grant store (#883): resource-keyed rows granting a user or
+client principal management authority over a vocabulary (level grants
+compared via the monotonic scope algebra) or bitflag access to a vault
+label. Administration routes in campus/auth/routes/grants.py; GET
+/grants/ is the "who can administer what" matrix query.
+
+Grant rows are plain dicts (the server's AccessGrant resource): id,
+grantee_type, grantee_id, resource_type, resource_id, bits, level,
+created_at. Vault rows carry bits (1=READ, 2=CREATE, 4=UPDATE,
+8=DELETE); management rows carry level (read < mod < write < admin).
+"""
+
+from ...interface import JsonDict, ResourceCollection
+
+
+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
+        resp = self.client.get(self.make_path(), query=query or None)
+        # Raise error if status code is not 2XX or 3XX
+        resp.raise_for_status()
+        return resp.json()
+
+    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
+        resp = self.client.get(self.make_path("check", end_slash=False), query=query)
+        resp.raise_for_status()
+        return bool(resp.json().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)
+        """
+        body: JsonDict = {
+            "grantee_type": grantee_type,
+            "grantee_id": grantee_id,
+            "resource_type": resource_type,
+        }
+        if resource_id:
+            body["resource_id"] = resource_id
+        if bits is not None:
+            body["bits"] = bits
+        if level is not None:
+            body["level"] = level
+        resp = self.client.post(self.make_path(), json=body)
+        resp.raise_for_status()
+        return resp.json()["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
+        """
+        body: JsonDict = {
+            "grantee_type": grantee_type,
+            "grantee_id": grantee_id,
+            "resource_type": resource_type,
+        }
+        if resource_id:
+            body["resource_id"] = resource_id
+        if bits is not None:
+            body["bits"] = bits
+        if level is not None:
+            body["level"] = level
+        resp = self.client.post(self.make_path("revoke", end_slash=False), json=body)
+        resp.raise_for_status()
+        return resp.json()["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())
+        """
+        resp = self.client.delete(self.make_path(grant_id, end_slash=False))
+        resp.raise_for_status()
diff --git a/tests/unit/test_auth_grants.py b/tests/unit/test_auth_grants.py
new file mode 100644
index 0000000..d379640
--- /dev/null
+++ b/tests/unit/test_auth_grants.py
@@ -0,0 +1,174 @@
+"""Contract tests for the auth grants resource (campus-cli#50).
+
+The grants surface (#883/#886) is the access-grant store's
+administration API: list (the "who can administer what" matrix),
+check, grant, revoke, and delete-by-id. The SDK is a thin pass-through
+— filters become query params, grant/revoke bodies carry exactly the
+caller's fields, and rows are plain dicts (no campus model exists for
+AccessGrant).
+"""
+
+import unittest
+from unittest.mock import Mock
+
+from campus_python.auth.v1 import AuthRoot
+
+# Grant row shape emitted by campus weekly
+# (campus/auth/routes/grants.py responses).
+GRANT_ROW = {
+    "id": "uid-grant-10131fe6",
+    "grantee_type": "user",
+    "grantee_id": "user@example.com",
+    "resource_type": "users",
+    "resource_id": "",
+    "bits": None,
+    "level": "write",
+    "created_at": "2026-10-10T15:10:00+00:00",
+}
+
+
+def make_auth() -> tuple[AuthRoot, Mock]:
+    """Create an AuthRoot backed by a mock JSON client."""
+    client = Mock()
+    return AuthRoot(json_client=client), client
+
+
+class TestGrantsList(unittest.TestCase):
+    """Grants.list() passes filters as query params."""
+
+    def setUp(self):
+        self.auth, self.client = make_auth()
+        response = Mock()
+        response.json.return_value = {"grants": [GRANT_ROW]}
+        self.client.get.return_value = response
+
+    def test_list_unfiltered_returns_envelope(self):
+        result = self.auth.grants.list()
+        args, kwargs = self.client.get.call_args
+        self.assertEqual(args[0], "/auth/v1/grants/")
+        self.assertIsNone(kwargs["query"])
+        self.assertEqual(result["grants"], [GRANT_ROW])
+
+    def test_list_forwards_filters(self):
+        self.auth.grants.list(
+            grantee_type="user",
+            grantee_id="user@example.com",
+            resource_type="users",
+        )
+        kwargs = self.client.get.call_args.kwargs
+        self.assertEqual(
+            kwargs["query"],
+            {
+                "grantee_type": "user",
+                "grantee_id": "user@example.com",
+                "resource_type": "users",
+            },
+        )
+
+
+class TestGrantsCheck(unittest.TestCase):
+    """Grants.check() returns the granted boolean."""
+
+    def setUp(self):
+        self.auth, self.client = make_auth()
+        response = Mock()
+        response.json.return_value = {"granted": True}
+        self.client.get.return_value = response
+
+    def test_check_sends_query_and_returns_bool(self):
+        granted = self.auth.grants.check(
+            grantee_type="user",
+            grantee_id="user@example.com",
+            resource_type="users",
+            level="write",
+        )
+        args, kwargs = self.client.get.call_args
+        self.assertEqual(args[0], "/auth/v1/grants/check")
+        self.assertEqual(
+            kwargs["query"],
+            {
+                "grantee_type": "user",
+                "grantee_id": "user@example.com",
+                "resource_type": "users",
+                "level": "write",
+            },
+        )
+        self.assertIs(granted, True)
+
+
+class TestGrantsGrantRevoke(unittest.TestCase):
+    """grant()/revoke() send exactly the caller's fields."""
+
+    def setUp(self):
+        self.auth, self.client = make_auth()
+
+    def test_grant_sends_level_row(self):
+        response = Mock()
+        response.json.return_value = {"grant": GRANT_ROW}
+        self.client.post.return_value = response
+        row = self.auth.grants.grant(
+            grantee_type="user",
+            grantee_id="user@example.com",
+            resource_type="users",
+            level="write",
+        )
+        args, kwargs = self.client.post.call_args
+        self.assertEqual(args[0], "/auth/v1/grants/")
+        self.assertEqual(
+            kwargs["json"],
+            {
+                "grantee_type": "user",
+                "grantee_id": "user@example.com",
+                "resource_type": "users",
+                "level": "write",
+            },
+        )
+        self.assertEqual(row["id"], GRANT_ROW["id"])
+
+    def test_grant_vault_sends_bits_and_label(self):
+        response = Mock()
+        response.json.return_value = {"grant": {**GRANT_ROW, "level": None, "bits": 1}}
+        self.client.post.return_value = response
+        self.auth.grants.grant(
+            grantee_type="client",
+            grantee_id="uid-client-abc",
+            resource_type="vault",
+            resource_id="email",
+            bits=1,
+        )
+        kwargs = self.client.post.call_args.kwargs
+        self.assertEqual(
+            kwargs["json"],
+            {
+                "grantee_type": "client",
+                "grantee_id": "uid-client-abc",
+                "resource_type": "vault",
+                "resource_id": "email",
+                "bits": 1,
+            },
+        )
+
+    def test_revoke_posts_to_revoke_path(self):
+        response = Mock()
+        response.json.return_value = {"grant": None}
+        self.client.post.return_value = response
+        row = self.auth.grants.revoke(
+            grantee_type="user",
+            grantee_id="user@example.com",
+            resource_type="users",
+            level="write",
+        )
+        args, _ = self.client.post.call_args
+        self.assertEqual(args[0], "/auth/v1/grants/revoke")
+        self.assertIsNone(row)
+
+    def test_delete_by_id(self):
+        response = Mock()
+        self.client.delete.return_value = response
+        self.auth.grants.delete("uid-grant-10131fe6")
+        args, _ = self.client.delete.call_args
+        self.assertEqual(args[0], "/auth/v1/grants/uid-grant-10131fe6")
+
+
+if __name__ == "__main__":
+    unittest.main()