wip2

tests/test_security.py

@@ -806,6 +806,204 @@ class TestOpenIDAuth:
         assert await oidc.extract(request) == "mytoken"
 
 
+class TestSyncValidators:
+    """Sync (non-async) validators — covers the sync path in _call_validator."""
+
+    def test_bearer_sync_validator(self):
+        def sync_validator(credential: str) -> dict:
+            if credential != VALID_TOKEN:
+                raise UnauthorizedError()
+            return {"user": "alice"}
+
+        bearer = BearerTokenAuth(sync_validator)
+
+        def setup(app: FastAPI):
+            @app.get("/me")
+            async def me(user=Security(bearer)):
+                return user
+
+        client = TestClient(_app(setup))
+        response = client.get("/me", headers={"Authorization": f"Bearer {VALID_TOKEN}"})
+        assert response.status_code == 200
+        assert response.json() == {"user": "alice"}
+
+    def test_sync_validator_via_authenticate(self):
+        """authenticate() with sync validator (MultiAuth path)."""
+
+        def sync_validator(credential: str) -> dict:
+            if credential != VALID_TOKEN:
+                raise UnauthorizedError()
+            return {"user": "alice"}
+
+        bearer = BearerTokenAuth(sync_validator)
+        multi = MultiAuth(bearer)
+
+        def setup(app: FastAPI):
+            @app.get("/me")
+            async def me(user=Security(multi)):
+                return user
+
+        client = TestClient(_app(setup))
+        response = client.get("/me", headers={"Authorization": f"Bearer {VALID_TOKEN}"})
+        assert response.status_code == 200
+        assert response.json() == {"user": "alice"}
+
+
+class TestOpenIDAuthEdgeCases:
+    DISCOVERY_URL = "https://accounts.example.com/.well-known/openid-configuration"
+
+    def test_non_bearer_authorization_returns_401(self):
+        """Authorization header present but not Bearer scheme."""
+        oidc = OpenIDAuth(self.DISCOVERY_URL, simple_validator)
+
+        def setup(app: FastAPI):
+            @app.get("/me")
+            async def me(user=Security(oidc)):
+                return user
+
+        client = TestClient(_app(setup))
+        response = client.get("/me", headers={"Authorization": "Basic dXNlcjpwYXNz"})
+        assert response.status_code == 401
+
+    def test_empty_bearer_token_returns_401(self):
+        """Authorization: Bearer with no token after the scheme prefix."""
+        oidc = OpenIDAuth(self.DISCOVERY_URL, simple_validator)
+
+        def setup(app: FastAPI):
+            @app.get("/me")
+            async def me(user=Security(oidc)):
+                return user
+
+        client = TestClient(_app(setup))
+        response = client.get("/me", headers={"Authorization": "Bearer "})
+        assert response.status_code == 401
+
+
+class TestCookieAuthSigned:
+    """CookieAuth with HMAC-SHA256 signed cookies (secret_key path)."""
+
+    SECRET = "test-hmac-secret"
+
+    def test_valid_signed_cookie_via_set_cookie(self):
+        """set_cookie signs the value; the signed cookie is verified on read."""
+        from fastapi import Response
+
+        auth = CookieAuth("session", cookie_validator, secret_key=self.SECRET)
+
+        def setup(app: FastAPI):
+            @app.get("/login")
+            async def login(response: Response):
+                auth.set_cookie(response, VALID_COOKIE)
+                return {"ok": True}
+
+            @app.get("/me")
+            async def me(user=Security(auth)):
+                return user
+
+        with TestClient(_app(setup)) as client:
+            client.get("/login")
+            response = client.get("/me")
+        assert response.status_code == 200
+        assert response.json() == {"session": VALID_COOKIE}
+
+    def test_tampered_signature_returns_401(self):
+        """A cookie whose HMAC signature has been modified is rejected."""
+        import base64 as _b64
+        import json as _json
+        import time as _time
+
+        auth = CookieAuth("session", cookie_validator, secret_key=self.SECRET)
+
+        def setup(app: FastAPI):
+            @app.get("/me")
+            async def me(user=Security(auth)):
+                return user
+
+        client = TestClient(_app(setup))
+        data = _b64.urlsafe_b64encode(
+            _json.dumps({"v": VALID_COOKIE, "exp": int(_time.time()) + 9999}).encode()
+        ).decode()
+        response = client.get("/me", cookies={"session": f"{data}.invalidsig"})
+        assert response.status_code == 401
+
+    def test_expired_signed_cookie_returns_401(self):
+        """A signed cookie past its expiry timestamp is rejected."""
+        import base64 as _b64
+        import hashlib as _hashlib
+        import hmac as _hmac
+        import json as _json
+        import time as _time
+
+        auth = CookieAuth("session", cookie_validator, secret_key=self.SECRET)
+
+        def setup(app: FastAPI):
+            @app.get("/me")
+            async def me(user=Security(auth)):
+                return user
+
+        client = TestClient(_app(setup))
+        data = _b64.urlsafe_b64encode(
+            _json.dumps({"v": VALID_COOKIE, "exp": int(_time.time()) - 1}).encode()
+        ).decode()
+        sig = _hmac.new(
+            self.SECRET.encode(), data.encode(), _hashlib.sha256
+        ).hexdigest()
+        response = client.get("/me", cookies={"session": f"{data}.{sig}"})
+        assert response.status_code == 401
+
+    def test_invalid_json_payload_returns_401(self):
+        """A signed cookie whose payload is not valid JSON is rejected."""
+        import base64 as _b64
+        import hashlib as _hashlib
+        import hmac as _hmac
+
+        auth = CookieAuth("session", cookie_validator, secret_key=self.SECRET)
+
+        def setup(app: FastAPI):
+            @app.get("/me")
+            async def me(user=Security(auth)):
+                return user
+
+        client = TestClient(_app(setup))
+        data = _b64.urlsafe_b64encode(b"not-valid-json").decode()
+        sig = _hmac.new(
+            self.SECRET.encode(), data.encode(), _hashlib.sha256
+        ).hexdigest()
+        response = client.get("/me", cookies={"session": f"{data}.{sig}"})
+        assert response.status_code == 401
+
+    def test_malformed_cookie_no_dot_returns_401(self):
+        """A signed cookie without the dot separator is rejected."""
+        auth = CookieAuth("session", cookie_validator, secret_key=self.SECRET)
+
+        def setup(app: FastAPI):
+            @app.get("/me")
+            async def me(user=Security(auth)):
+                return user
+
+        client = TestClient(_app(setup))
+        response = client.get("/me", cookies={"session": "nodothere"})
+        assert response.status_code == 401
+
+    def test_set_cookie_without_secret(self):
+        """set_cookie without secret_key writes the raw value."""
+        from starlette.responses import Response as StarletteResponse
+
+        auth = CookieAuth("session", cookie_validator)
+        response = StarletteResponse()
+        auth.set_cookie(response, "rawvalue")
+        assert "session=rawvalue" in response.headers["set-cookie"]
+
+    def test_delete_cookie(self):
+        """delete_cookie produces a Set-Cookie header that clears the session."""
+        from starlette.responses import Response as StarletteResponse
+
+        auth = CookieAuth("session", cookie_validator)
+        response = StarletteResponse()
+        auth.delete_cookie(response)
+        assert "session" in response.headers["set-cookie"]
+
+
 # Minimal concrete subclass used only in tests below.
 class _HeaderAuth(AuthSource):
     """Reads a custom X-Token header — no FastAPI security scheme."""