Fix cache scope validation for org-level tokens

src/nimbus/common/security.py

@@ -190,21 +190,20 @@ def validate_cache_scope(token: CacheToken, operation: str, org_id: int, *, repo
     # Legacy tokens with simple scopes
     scopes = [s.strip() for s in token.scope.split(",") if s.strip()]
 
-    if repo:
-        sanitized = repo.strip("/")
-        required_repo_scope = f"{operation}:org-{org_id}/{sanitized}"
-        if required_repo_scope in scopes:
-            return True
-
     if token.scope == "read_write":
         return True
     if token.scope == "read" and operation == "pull":
         return True
     if token.scope == "write" and operation == "push":
         return True
 
+    required_scope = f"{operation}:org-{org_id}"
+    if required_scope in scopes:
+        return True
+
     if repo:
-        return False
+        sanitized = repo.strip("/")
+        required_repo_scope = f"{operation}:org-{org_id}/{sanitized}"
+        return required_repo_scope in scopes
 
-    required_scope = f"{operation}:org-{org_id}"
-    return required_scope in scopes
+    return False

tests/conftest.py

@@ -61,3 +61,24 @@ def generate_session_token(self, *_args, **_kwargs):
 saml_module.SamlAuthenticator = _DummySamlAuthenticator  # type: ignore[attr-defined]
 saml_module.SamlValidationError = Exception  # type: ignore[attr-defined]
 sys.modules.setdefault("nimbus.control_plane.saml", saml_module)
+sys.modules.setdefault("src.nimbus.control_plane.saml", saml_module)
+
+if "saml2" not in sys.modules:
+    saml2_module = ModuleType("saml2")
+    saml2_module.BINDING_HTTP_POST = "post"
+    saml2_module.BINDING_HTTP_REDIRECT = "redirect"
+    sys.modules["saml2"] = saml2_module
+
+    saml2_client_module = ModuleType("saml2.client")
+    saml2_client_module.Saml2Client = SimpleNamespace
+    sys.modules["saml2.client"] = saml2_client_module
+
+    saml2_config_module = ModuleType("saml2.config")
+    saml2_config_module.Config = SimpleNamespace
+    sys.modules["saml2.config"] = saml2_config_module
+
+    saml2_metadata_module = ModuleType("saml2.metadata")
+    saml2_metadata_module.entity_descriptor = lambda *args, **kwargs: SimpleNamespace(
+        to_string=lambda: b"<EntityDescriptor/>"
+    )
+    sys.modules["saml2.metadata"] = saml2_metadata_module

tests/test_docker_cache.py

@@ -143,10 +143,11 @@ async def test_manifest_roundtrip(tmp_path: Path, monkeypatch: pytest.MonkeyPatc
         assert head_resp.headers["Docker-Content-Digest"] == manifest_digest
         assert head_resp.headers["Content-Length"] == str(len(manifest_bytes))
 
-        # repo-scoped denial
+        # org-wide scope access should succeed without repo-specific suffix
         restricted_headers = await _auth_headers(secret, scope="pull:org-1,push:org-1")
-        deny_resp = await client.get("/v2/org-1/demo/manifests/latest", headers=restricted_headers)
-        assert deny_resp.status_code == 404
+        allow_resp = await client.get("/v2/org-1/demo/manifests/latest", headers=restricted_headers)
+        assert allow_resp.status_code == 200
+        assert json.loads(allow_resp.content) == manifest
     finally:
         await client.aclose()
         await lifespan.__aexit__(None, None, None)