omnigent-ai--omnigent
910 行
34 KiB
Python
910 行
34 KiB
Python
"""Unit tests for OIDC authentication: OIDCConfig, UnifiedAuthProvider,
|
|
create_auth_provider factory, PKCE helpers, and session cookie utilities.
|
|
|
|
Tests mirror the source at ``omnigent/server/oidc.py`` and
|
|
``omnigent/server/auth.py``.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
from unittest.mock import MagicMock
|
|
|
|
import jwt
|
|
import pytest
|
|
|
|
from omnigent.server.auth import (
|
|
RESERVED_USER_LOCAL,
|
|
UnifiedAuthProvider,
|
|
create_auth_provider,
|
|
resolve_auth_header,
|
|
resolve_auth_header_strip_prefix,
|
|
)
|
|
from omnigent.server.oidc import (
|
|
OIDCConfig,
|
|
derive_code_challenge,
|
|
generate_code_verifier,
|
|
hmac_digest,
|
|
mint_session_cookie,
|
|
mint_session_token,
|
|
)
|
|
|
|
# ── PKCE helpers ─────────────────────────────────────────────────
|
|
|
|
|
|
def test_generate_code_verifier_length() -> None:
|
|
"""Code verifier is within the RFC 7636 length bounds (43–128 chars).
|
|
|
|
A verifier outside this range would be rejected by compliant
|
|
IdPs during the token exchange.
|
|
"""
|
|
verifier = generate_code_verifier()
|
|
# RFC 7636 requires 43–128 characters.
|
|
assert 43 <= len(verifier) <= 128, (
|
|
f"Code verifier length {len(verifier)} is outside RFC 7636 bounds (43–128)."
|
|
)
|
|
|
|
|
|
def test_derive_code_challenge_is_deterministic() -> None:
|
|
"""Same verifier always produces the same S256 challenge.
|
|
|
|
Non-deterministic challenges would fail the PKCE verification
|
|
at the IdP's token endpoint.
|
|
"""
|
|
verifier = "test_verifier_12345678901234567890123456789012345"
|
|
c1 = derive_code_challenge(verifier)
|
|
c2 = derive_code_challenge(verifier)
|
|
assert c1 == c2, "Code challenge must be deterministic."
|
|
# Must be base64url without padding.
|
|
assert "=" not in c1, "Code challenge must not have base64 padding."
|
|
assert "+" not in c1, "Code challenge must use base64url, not base64."
|
|
|
|
|
|
def test_derive_code_challenge_differs_for_different_verifiers() -> None:
|
|
"""Different verifiers produce different challenges.
|
|
|
|
If they didn't, PKCE would provide no security — any
|
|
code_verifier would match any challenge.
|
|
"""
|
|
v1 = generate_code_verifier()
|
|
v2 = generate_code_verifier()
|
|
assert derive_code_challenge(v1) != derive_code_challenge(v2)
|
|
|
|
|
|
# ── Session cookie minting / validation ──────────────────────────
|
|
|
|
|
|
_TEST_SECRET = b"a" * 32 # 32 bytes — minimum valid secret
|
|
|
|
|
|
def test_mint_session_cookie_produces_valid_jwt() -> None:
|
|
"""Minted cookie is a valid HS256 JWT with the expected claims.
|
|
|
|
If the cookie fails to decode, the OIDCAuthProvider would
|
|
reject every request and return None (→ 401 for all users).
|
|
"""
|
|
token = mint_session_cookie(
|
|
user_id="alice@example.com",
|
|
cookie_secret=_TEST_SECRET,
|
|
ttl_hours=8,
|
|
provider="google",
|
|
)
|
|
payload = jwt.decode(token, _TEST_SECRET, algorithms=["HS256"])
|
|
assert payload["sub"] == "alice@example.com"
|
|
assert payload["provider"] == "google"
|
|
# exp should be ~8 hours from now.
|
|
assert payload["exp"] > time.time()
|
|
assert payload["exp"] <= time.time() + (8 * 3600) + 5
|
|
|
|
|
|
def test_mint_session_cookie_rejected_with_wrong_secret() -> None:
|
|
"""Cookie signed with one secret fails validation with another.
|
|
|
|
This is the core security property: a stolen cookie from one
|
|
deployment cannot be used on another deployment with a different
|
|
secret.
|
|
"""
|
|
token = mint_session_cookie(
|
|
user_id="alice@example.com",
|
|
cookie_secret=_TEST_SECRET,
|
|
ttl_hours=8,
|
|
provider="google",
|
|
)
|
|
wrong_secret = b"b" * 32
|
|
with pytest.raises(jwt.InvalidSignatureError):
|
|
jwt.decode(token, wrong_secret, algorithms=["HS256"])
|
|
|
|
|
|
def test_mint_session_token_produces_valid_jwt_with_seconds_ttl() -> None:
|
|
"""The seconds-granularity primitive mints the same HS256 claim shape.
|
|
|
|
A managed runner needs a sub-hour owner token; ``mint_session_token``
|
|
is the seconds-based core the hours-only ``mint_session_cookie``
|
|
cannot express. The claims must match so the same validator accepts
|
|
either.
|
|
"""
|
|
token = mint_session_token(
|
|
user_id="alice@example.com",
|
|
cookie_secret=_TEST_SECRET,
|
|
ttl_seconds=120,
|
|
provider="accounts",
|
|
)
|
|
payload = jwt.decode(token, _TEST_SECRET, algorithms=["HS256"])
|
|
assert payload["sub"] == "alice@example.com"
|
|
assert payload["provider"] == "accounts"
|
|
# exp is ~120 seconds out, not hours.
|
|
assert time.time() < payload["exp"] <= time.time() + 120 + 5
|
|
|
|
|
|
def test_mint_session_token_short_ttl_expires() -> None:
|
|
"""A past TTL yields a token the HS256 validator rejects as expired.
|
|
|
|
This is the property that removes the fixed session-length cap: the
|
|
minted owner token genuinely expires, so the runner must (and does)
|
|
re-mint — rather than holding one static long-lived credential.
|
|
"""
|
|
token = mint_session_token(
|
|
user_id="alice@example.com",
|
|
cookie_secret=_TEST_SECRET,
|
|
ttl_seconds=-1,
|
|
provider="accounts",
|
|
)
|
|
with pytest.raises(jwt.ExpiredSignatureError):
|
|
jwt.decode(token, _TEST_SECRET, algorithms=["HS256"])
|
|
|
|
|
|
def test_hmac_digest_is_deterministic() -> None:
|
|
"""Same token + secret always produces the same digest.
|
|
|
|
The credential cache in UnifiedAuthProvider uses this as a
|
|
cache key — non-deterministic digests would cause 100% cache
|
|
misses.
|
|
"""
|
|
d1 = hmac_digest("my-token", _TEST_SECRET)
|
|
d2 = hmac_digest("my-token", _TEST_SECRET)
|
|
assert d1 == d2
|
|
|
|
|
|
def test_hmac_digest_differs_for_different_tokens() -> None:
|
|
"""Different tokens produce different digests.
|
|
|
|
If they didn't, the cache would return the wrong user for
|
|
a different cookie.
|
|
"""
|
|
d1 = hmac_digest("token-a", _TEST_SECRET)
|
|
d2 = hmac_digest("token-b", _TEST_SECRET)
|
|
assert d1 != d2
|
|
|
|
|
|
# ── UnifiedAuthProvider (header source) ──────────────────────────
|
|
|
|
|
|
def _mock_request(
|
|
headers: dict[str, str] | None = None,
|
|
cookies: dict[str, str] | None = None,
|
|
) -> MagicMock:
|
|
"""Build a minimal mock HTTPConnection.
|
|
|
|
MagicMock is acceptable here: we only need ``request.headers``
|
|
(a dict-like) and ``request.cookies`` (a dict-like), and
|
|
``HTTPConnection`` is a complex ASGI object that cannot be
|
|
trivially constructed without a real scope.
|
|
|
|
:param headers: Request headers dict.
|
|
:param cookies: Request cookies dict.
|
|
:returns: A mock with ``.headers`` and ``.cookies`` set.
|
|
"""
|
|
mock = MagicMock()
|
|
mock.headers = headers or {}
|
|
mock.cookies = cookies or {}
|
|
return mock
|
|
|
|
|
|
def test_header_source_returns_email_from_header() -> None:
|
|
"""Header source extracts user ID from X-Forwarded-Email.
|
|
|
|
This is the primary code path for Databricks Apps deployments
|
|
where the proxy injects the header.
|
|
"""
|
|
provider = UnifiedAuthProvider(source="header")
|
|
request = _mock_request(headers={"X-Forwarded-Email": "alice@example.com"})
|
|
assert provider.get_user_id(request) == "alice@example.com"
|
|
|
|
|
|
def test_header_source_rejects_missing_header(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""Missing header is rejected (``None`` → 401) by default.
|
|
|
|
A missing or proxy-dropped ``X-Forwarded-Email``
|
|
must fail closed. If this returned ``"local"`` instead, every
|
|
unauthenticated request would share one identity with OWNER
|
|
access to every other unauthenticated user's sessions.
|
|
"""
|
|
# The provider resolves OMNIGENT_LOCAL_SINGLE_USER at
|
|
# construction; clear it so an ambient value (e.g. a dev shell
|
|
# that ran a local server) can't flip this test to the
|
|
# single-user fallback path.
|
|
monkeypatch.delenv("OMNIGENT_LOCAL_SINGLE_USER", raising=False)
|
|
provider = UnifiedAuthProvider(source="header")
|
|
request = _mock_request()
|
|
assert provider.get_user_id(request) is None
|
|
|
|
|
|
def test_header_source_local_single_user_falls_back_to_local() -> None:
|
|
"""Explicit single-user runtime keeps the ``"local"`` fallback.
|
|
|
|
``local_single_user=True`` models a server spawned with
|
|
``OMNIGENT_LOCAL_SINGLE_USER=1`` (the managed local spawn
|
|
paths) — its only user IS the local user, and no proxy exists
|
|
to inject identity.
|
|
"""
|
|
provider = UnifiedAuthProvider(source="header", local_single_user=True)
|
|
request = _mock_request()
|
|
assert provider.get_user_id(request) == RESERVED_USER_LOCAL
|
|
|
|
|
|
def test_header_source_resolves_single_user_from_env(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""``local_single_user=None`` resolves from the env at construction.
|
|
|
|
The managed local spawn paths configure single-user mode via the
|
|
``OMNIGENT_LOCAL_SINGLE_USER=1`` env var, not a constructor
|
|
argument — this pins the env-resolution path they rely on.
|
|
"""
|
|
monkeypatch.setenv("OMNIGENT_LOCAL_SINGLE_USER", "1")
|
|
provider = UnifiedAuthProvider(source="header")
|
|
request = _mock_request()
|
|
assert provider.get_user_id(request) == RESERVED_USER_LOCAL
|
|
|
|
|
|
def test_header_source_single_user_still_honors_header() -> None:
|
|
"""A present header wins over the single-user fallback.
|
|
|
|
Even on a single-user local runtime, a proxy-injected identity
|
|
must be honored (and reserved names rejected) — the fallback
|
|
only covers the header-absent case.
|
|
"""
|
|
provider = UnifiedAuthProvider(source="header", local_single_user=True)
|
|
request = _mock_request(headers={"X-Forwarded-Email": "alice@example.com"})
|
|
assert provider.get_user_id(request) == "alice@example.com"
|
|
|
|
|
|
@pytest.mark.parametrize("reserved", ["local", "__public__"])
|
|
def test_header_source_rejects_reserved_names(
|
|
reserved: str,
|
|
) -> None:
|
|
"""Reserved names are rejected in header mode.
|
|
|
|
If accepted, a client could impersonate the admin or public
|
|
sentinel by sending ``X-Forwarded-Email: local``.
|
|
"""
|
|
provider = UnifiedAuthProvider(source="header")
|
|
request = _mock_request(headers={"X-Forwarded-Email": reserved})
|
|
assert provider.get_user_id(request) is None
|
|
|
|
|
|
def test_header_source_login_url_is_none() -> None:
|
|
"""Header-only provider has no login URL.
|
|
|
|
The frontend uses this to know it should NOT redirect to a
|
|
login page when /v1/me returns null.
|
|
"""
|
|
provider = UnifiedAuthProvider(source="header")
|
|
assert provider.login_url is None
|
|
|
|
|
|
def test_header_source_reads_custom_header_name() -> None:
|
|
"""An explicit ``header_name`` reads identity from that header.
|
|
|
|
Models a deploy behind Cloudflare Access, which authenticates
|
|
with ``Cf-Access-Authenticated-User-Email`` rather than the
|
|
``X-Forwarded-Email`` default (issue #877).
|
|
"""
|
|
provider = UnifiedAuthProvider(
|
|
source="header",
|
|
header_name="Cf-Access-Authenticated-User-Email",
|
|
)
|
|
request = _mock_request(
|
|
headers={"Cf-Access-Authenticated-User-Email": "alice@example.com"},
|
|
)
|
|
assert provider.get_user_id(request) == "alice@example.com"
|
|
|
|
|
|
def test_header_source_custom_header_ignores_default_header() -> None:
|
|
"""With a custom header set, the default header no longer authenticates.
|
|
|
|
This is the security-relevant half of the override: an operator
|
|
who points the server at the proxy's header must NOT also leave
|
|
the old ``X-Forwarded-Email`` as a second accepted identity, or a
|
|
client could spoof identity through the header the proxy doesn't
|
|
strip.
|
|
"""
|
|
provider = UnifiedAuthProvider(
|
|
source="header",
|
|
header_name="Cf-Access-Authenticated-User-Email",
|
|
local_single_user=False,
|
|
)
|
|
request = _mock_request(headers={"X-Forwarded-Email": "attacker@example.com"})
|
|
assert provider.get_user_id(request) is None
|
|
|
|
|
|
def test_header_source_resolves_header_name_from_env(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""``header_name=None`` resolves from ``OMNIGENT_AUTH_HEADER``.
|
|
|
|
The deploy path configures the header name via env var, not a
|
|
constructor argument — this pins the env-resolution path.
|
|
"""
|
|
monkeypatch.setenv("OMNIGENT_AUTH_HEADER", "Cf-Access-Authenticated-User-Email")
|
|
provider = UnifiedAuthProvider(source="header")
|
|
request = _mock_request(
|
|
headers={"Cf-Access-Authenticated-User-Email": "alice@example.com"},
|
|
)
|
|
assert provider.get_user_id(request) == "alice@example.com"
|
|
|
|
|
|
def test_resolve_auth_header_defaults_to_x_forwarded_email(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""Unset/empty ``OMNIGENT_AUTH_HEADER`` falls back to the default.
|
|
|
|
The override must be strictly additive: the overwhelming majority
|
|
of header-mode deploys (and every existing test) rely on
|
|
``X-Forwarded-Email`` being the default.
|
|
"""
|
|
monkeypatch.delenv("OMNIGENT_AUTH_HEADER", raising=False)
|
|
assert resolve_auth_header() == "X-Forwarded-Email"
|
|
monkeypatch.setenv("OMNIGENT_AUTH_HEADER", " ")
|
|
assert resolve_auth_header() == "X-Forwarded-Email"
|
|
|
|
|
|
# ── UnifiedAuthProvider (header source: Google IAP prefix strip) ──
|
|
|
|
# Google IAP injects its identity in ``X-Goog-Authenticated-User-Email``
|
|
# namespaced as ``accounts.google.com:<email>``. The configured strip
|
|
# prefix turns that into the bare email used everywhere else.
|
|
_IAP_HEADER = "X-Goog-Authenticated-User-Email"
|
|
_IAP_PREFIX = "accounts.google.com:"
|
|
|
|
|
|
def test_header_source_strips_configured_prefix() -> None:
|
|
"""Google IAP: the ``accounts.google.com:`` prefix is stripped.
|
|
|
|
IAP forwards ``accounts.google.com:user@example.com``. Without
|
|
stripping, the identity would carry the namespace prefix and never
|
|
match the bare email used for ownership/sharing — so a deploy behind
|
|
IAP could not be addressed by its real identity at all.
|
|
"""
|
|
provider = UnifiedAuthProvider(
|
|
source="header",
|
|
header_name=_IAP_HEADER,
|
|
header_strip_prefix=_IAP_PREFIX,
|
|
)
|
|
request = _mock_request(headers={_IAP_HEADER: f"{_IAP_PREFIX}alice@example.com"})
|
|
assert provider.get_user_id(request) == "alice@example.com"
|
|
|
|
|
|
def test_header_source_strip_prefix_absent_passes_value_through() -> None:
|
|
"""A value lacking the prefix is used unchanged.
|
|
|
|
``str.removeprefix`` is a no-op when the prefix is absent, so a
|
|
proxy value that is already bare still authenticates — the strip is
|
|
purely additive and never corrupts a non-namespaced identity.
|
|
"""
|
|
provider = UnifiedAuthProvider(
|
|
source="header",
|
|
header_name=_IAP_HEADER,
|
|
header_strip_prefix=_IAP_PREFIX,
|
|
)
|
|
request = _mock_request(headers={_IAP_HEADER: "alice@example.com"})
|
|
assert provider.get_user_id(request) == "alice@example.com"
|
|
|
|
|
|
def test_header_source_rejects_value_that_is_only_the_prefix() -> None:
|
|
"""A header carrying only the prefix (empty after strip) fails closed.
|
|
|
|
``accounts.google.com:`` with no email is malformed; it must reject
|
|
(``None`` → 401), never authenticate as an empty-string identity
|
|
that requests could then share.
|
|
"""
|
|
provider = UnifiedAuthProvider(
|
|
source="header",
|
|
header_name=_IAP_HEADER,
|
|
header_strip_prefix=_IAP_PREFIX,
|
|
local_single_user=False,
|
|
)
|
|
request = _mock_request(headers={_IAP_HEADER: _IAP_PREFIX})
|
|
assert provider.get_user_id(request) is None
|
|
|
|
|
|
def test_header_source_rejects_reserved_name_behind_prefix() -> None:
|
|
"""Reserved-name rejection applies AFTER stripping.
|
|
|
|
Otherwise ``accounts.google.com:local`` would slip past the
|
|
reserved check (it isn't literally ``"local"``) and then strip down
|
|
to the reserved ``"local"`` sentinel — letting a client impersonate
|
|
it through the namespaced header.
|
|
"""
|
|
provider = UnifiedAuthProvider(
|
|
source="header",
|
|
header_name=_IAP_HEADER,
|
|
header_strip_prefix=_IAP_PREFIX,
|
|
)
|
|
request = _mock_request(headers={_IAP_HEADER: f"{_IAP_PREFIX}local"})
|
|
assert provider.get_user_id(request) is None
|
|
|
|
|
|
def test_header_source_resolves_strip_prefix_from_env(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""``header_strip_prefix=None`` resolves from the env var.
|
|
|
|
The deploy path configures the prefix via
|
|
``OMNIGENT_AUTH_HEADER_STRIP_PREFIX``, not a constructor argument —
|
|
this pins the env-resolution path IAP deployments rely on.
|
|
"""
|
|
monkeypatch.setenv("OMNIGENT_AUTH_HEADER", _IAP_HEADER)
|
|
monkeypatch.setenv("OMNIGENT_AUTH_HEADER_STRIP_PREFIX", _IAP_PREFIX)
|
|
provider = UnifiedAuthProvider(source="header")
|
|
request = _mock_request(headers={_IAP_HEADER: f"{_IAP_PREFIX}alice@example.com"})
|
|
assert provider.get_user_id(request) == "alice@example.com"
|
|
|
|
|
|
def test_resolve_auth_header_strip_prefix_defaults_to_empty(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""Unset/whitespace ``OMNIGENT_AUTH_HEADER_STRIP_PREFIX`` → ``""``.
|
|
|
|
The default must strip nothing so every existing header-mode deploy
|
|
(none of which set this) keeps passing the header value through
|
|
verbatim.
|
|
"""
|
|
monkeypatch.delenv("OMNIGENT_AUTH_HEADER_STRIP_PREFIX", raising=False)
|
|
assert resolve_auth_header_strip_prefix() == ""
|
|
monkeypatch.setenv("OMNIGENT_AUTH_HEADER_STRIP_PREFIX", " ")
|
|
assert resolve_auth_header_strip_prefix() == ""
|
|
|
|
|
|
# ── UnifiedAuthProvider (oidc source) ────────────────────────────
|
|
|
|
|
|
def _make_oidc_config() -> OIDCConfig:
|
|
"""Build a minimal OIDCConfig for testing cookie validation."""
|
|
return OIDCConfig(
|
|
issuer="https://accounts.google.com",
|
|
client_id="test-client",
|
|
client_secret="test-secret",
|
|
redirect_uri="https://app.example.com/auth/callback",
|
|
cookie_secret=_TEST_SECRET,
|
|
scopes="openid email profile",
|
|
session_ttl_hours=8,
|
|
logout_redirect_uri=None,
|
|
allowed_domains=None,
|
|
provider_type="oidc",
|
|
authorization_endpoint="https://accounts.google.com/o/oauth2/v2/auth",
|
|
token_endpoint="https://oauth2.googleapis.com/token",
|
|
jwks_uri="https://www.googleapis.com/oauth2/v3/certs",
|
|
userinfo_endpoint=None,
|
|
allow_invites=False,
|
|
)
|
|
|
|
|
|
def test_oidc_source_reads_valid_cookie() -> None:
|
|
"""OIDC source extracts user ID from a valid session cookie.
|
|
|
|
This is the primary code path for standalone OIDC deployments
|
|
where the server minted the cookie after a login flow.
|
|
"""
|
|
config = _make_oidc_config()
|
|
provider = UnifiedAuthProvider(source="oidc", oidc_config=config)
|
|
|
|
token = mint_session_cookie(
|
|
user_id="alice@example.com",
|
|
cookie_secret=_TEST_SECRET,
|
|
ttl_hours=8,
|
|
provider="google",
|
|
)
|
|
request = _mock_request(cookies={config.session_cookie_name: token})
|
|
assert provider.get_user_id(request) == "alice@example.com"
|
|
|
|
|
|
def test_oidc_source_returns_none_for_missing_cookie() -> None:
|
|
"""OIDC source returns None when no session cookie is present.
|
|
|
|
This triggers the 401 + login_url response from /v1/me,
|
|
which the frontend uses to redirect to the login page.
|
|
"""
|
|
config = _make_oidc_config()
|
|
provider = UnifiedAuthProvider(source="oidc", oidc_config=config)
|
|
request = _mock_request()
|
|
assert provider.get_user_id(request) is None
|
|
|
|
|
|
def test_oidc_source_returns_none_for_tampered_cookie() -> None:
|
|
"""OIDC source rejects a cookie signed with a different secret.
|
|
|
|
If it accepted tampered cookies, an attacker could forge
|
|
arbitrary session cookies.
|
|
"""
|
|
config = _make_oidc_config()
|
|
provider = UnifiedAuthProvider(source="oidc", oidc_config=config)
|
|
|
|
# Sign with a different secret.
|
|
wrong_secret = b"b" * 32
|
|
token = mint_session_cookie(
|
|
user_id="alice@example.com",
|
|
cookie_secret=wrong_secret,
|
|
ttl_hours=8,
|
|
provider="google",
|
|
)
|
|
request = _mock_request(cookies={config.session_cookie_name: token})
|
|
assert provider.get_user_id(request) is None
|
|
|
|
|
|
def test_oidc_source_returns_none_for_expired_cookie() -> None:
|
|
"""OIDC source rejects an expired session cookie.
|
|
|
|
Expired cookies must force re-authentication through the
|
|
login flow.
|
|
"""
|
|
config = _make_oidc_config()
|
|
provider = UnifiedAuthProvider(source="oidc", oidc_config=config)
|
|
|
|
# Mint a cookie that expired 1 second ago.
|
|
payload = {
|
|
"sub": "alice@example.com",
|
|
"iat": int(time.time()) - 3600,
|
|
"exp": int(time.time()) - 1,
|
|
"provider": "google",
|
|
}
|
|
token = jwt.encode(payload, _TEST_SECRET, algorithm="HS256")
|
|
request = _mock_request(cookies={config.session_cookie_name: token})
|
|
assert provider.get_user_id(request) is None
|
|
|
|
|
|
@pytest.mark.parametrize("reserved", ["local", "__public__"])
|
|
def test_oidc_source_rejects_reserved_sub_claims(reserved: str) -> None:
|
|
"""OIDC source rejects cookies with reserved user names in sub.
|
|
|
|
If a cookie somehow contained "local" or "__public__" as the
|
|
sub claim, it must not be accepted — these are internal
|
|
sentinel values.
|
|
"""
|
|
config = _make_oidc_config()
|
|
provider = UnifiedAuthProvider(source="oidc", oidc_config=config)
|
|
|
|
token = mint_session_cookie(
|
|
user_id=reserved,
|
|
cookie_secret=_TEST_SECRET,
|
|
ttl_hours=8,
|
|
provider="google",
|
|
)
|
|
request = _mock_request(cookies={config.session_cookie_name: token})
|
|
assert provider.get_user_id(request) is None
|
|
|
|
|
|
def test_oidc_source_caches_validated_cookie() -> None:
|
|
"""OIDC source caches a validated cookie to avoid repeated JWT decode.
|
|
|
|
The TTL credential cache prevents decoding the same cookie on
|
|
every request. If caching is broken, JWT decode runs on every
|
|
request — a significant performance regression under load.
|
|
"""
|
|
config = _make_oidc_config()
|
|
provider = UnifiedAuthProvider(source="oidc", oidc_config=config)
|
|
|
|
token = mint_session_cookie(
|
|
user_id="alice@example.com",
|
|
cookie_secret=_TEST_SECRET,
|
|
ttl_hours=8,
|
|
provider="google",
|
|
)
|
|
request = _mock_request(cookies={config.session_cookie_name: token})
|
|
|
|
# First call populates the cache.
|
|
assert provider.get_user_id(request) == "alice@example.com"
|
|
# Cache should now have one entry.
|
|
assert len(provider._cookie_cache) == 1
|
|
|
|
# Second call should use the cache (same result, no new entry).
|
|
assert provider.get_user_id(request) == "alice@example.com"
|
|
assert len(provider._cookie_cache) == 1
|
|
|
|
|
|
def test_oidc_source_accepts_bearer_token() -> None:
|
|
"""OIDC source accepts a session JWT via Authorization: Bearer header.
|
|
|
|
This is the code path used by ``omnigent run --server`` after
|
|
``omnigent login`` stores a token. The CLI sends the JWT as
|
|
a Bearer token instead of a cookie.
|
|
"""
|
|
config = _make_oidc_config()
|
|
provider = UnifiedAuthProvider(source="oidc", oidc_config=config)
|
|
|
|
token = mint_session_cookie(
|
|
user_id="alice@example.com",
|
|
cookie_secret=_TEST_SECRET,
|
|
ttl_hours=8,
|
|
provider="google",
|
|
)
|
|
# No cookie, but Bearer header present.
|
|
request = _mock_request(
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
assert provider.get_user_id(request) == "alice@example.com"
|
|
|
|
|
|
def test_oidc_source_cookie_takes_precedence_over_bearer() -> None:
|
|
"""When both cookie and Bearer token are present, cookie wins.
|
|
|
|
The cookie is set by the browser login flow and is more
|
|
authoritative. The Bearer token is a fallback for CLI clients.
|
|
"""
|
|
config = _make_oidc_config()
|
|
provider = UnifiedAuthProvider(source="oidc", oidc_config=config)
|
|
|
|
cookie_token = mint_session_cookie(
|
|
user_id="cookie-user@example.com",
|
|
cookie_secret=_TEST_SECRET,
|
|
ttl_hours=8,
|
|
provider="google",
|
|
)
|
|
bearer_token = mint_session_cookie(
|
|
user_id="bearer-user@example.com",
|
|
cookie_secret=_TEST_SECRET,
|
|
ttl_hours=8,
|
|
provider="google",
|
|
)
|
|
request = _mock_request(
|
|
cookies={config.session_cookie_name: cookie_token},
|
|
headers={"Authorization": f"Bearer {bearer_token}"},
|
|
)
|
|
# Cookie user wins.
|
|
assert provider.get_user_id(request) == "cookie-user@example.com"
|
|
|
|
|
|
def test_oidc_source_rejects_invalid_bearer_token() -> None:
|
|
"""OIDC source rejects a Bearer token signed with a wrong secret.
|
|
|
|
An attacker sending a forged Bearer token must not gain access.
|
|
"""
|
|
config = _make_oidc_config()
|
|
provider = UnifiedAuthProvider(source="oidc", oidc_config=config)
|
|
|
|
wrong_secret = b"b" * 32
|
|
token = mint_session_cookie(
|
|
user_id="alice@example.com",
|
|
cookie_secret=wrong_secret,
|
|
ttl_hours=8,
|
|
provider="google",
|
|
)
|
|
request = _mock_request(
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
assert provider.get_user_id(request) is None
|
|
|
|
|
|
def test_oidc_source_login_url() -> None:
|
|
"""OIDC provider exposes /auth/login as the login URL.
|
|
|
|
The /v1/me endpoint uses this to return 401 with a login_url
|
|
so the frontend knows where to redirect.
|
|
"""
|
|
config = _make_oidc_config()
|
|
provider = UnifiedAuthProvider(source="oidc", oidc_config=config)
|
|
assert provider.login_url == "/auth/login"
|
|
|
|
|
|
# ── create_auth_provider factory ─────────────────────────────────
|
|
|
|
|
|
def test_factory_returns_header_provider_explicit_zero_config(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""Explicit OMNIGENT_AUTH_PROVIDER=header returns a header provider.
|
|
|
|
This is the zero-config path for Databricks Apps and any deploy
|
|
behind a proxy that injects ``X-Forwarded-Email``. Header is the
|
|
env-unset default, but proxy-fronted deploys set the value
|
|
explicitly so an ambient ``OMNIGENT_AUTH_ENABLED=1`` can't
|
|
flip them into accounts (or oidc) mode.
|
|
"""
|
|
monkeypatch.setenv("OMNIGENT_AUTH_PROVIDER", "header")
|
|
provider = create_auth_provider()
|
|
assert isinstance(provider, UnifiedAuthProvider)
|
|
assert provider._source == "header"
|
|
assert provider.login_url is None
|
|
|
|
|
|
def test_factory_returns_header_provider_explicit(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""Explicit OMNIGENT_AUTH_PROVIDER=header returns a header provider."""
|
|
monkeypatch.setenv("OMNIGENT_AUTH_PROVIDER", "header")
|
|
provider = create_auth_provider()
|
|
assert isinstance(provider, UnifiedAuthProvider)
|
|
assert provider._source == "header"
|
|
|
|
|
|
def test_factory_rejects_unknown_source(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""Unknown provider value raises RuntimeError at startup.
|
|
|
|
This is the fail-loud behavior: a typo in the env var
|
|
surfaces immediately, not at the first request.
|
|
"""
|
|
monkeypatch.setenv("OMNIGENT_AUTH_PROVIDER", "saml")
|
|
with pytest.raises(RuntimeError, match="Unknown OMNIGENT_AUTH_PROVIDER"):
|
|
create_auth_provider()
|
|
|
|
|
|
def test_factory_oidc_fails_without_config(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""OIDC provider fails at startup when required env vars are missing.
|
|
|
|
This is the fail-loud startup validation: missing OIDC config
|
|
surfaces as a clear error before the server accepts connections.
|
|
"""
|
|
monkeypatch.setenv("OMNIGENT_AUTH_PROVIDER", "oidc")
|
|
# Don't set any OMNIGENT_OIDC_* vars.
|
|
monkeypatch.delenv("OMNIGENT_OIDC_ISSUER", raising=False)
|
|
monkeypatch.delenv("OMNIGENT_OIDC_CLIENT_ID", raising=False)
|
|
monkeypatch.delenv("OMNIGENT_OIDC_CLIENT_SECRET", raising=False)
|
|
monkeypatch.delenv("OMNIGENT_OIDC_REDIRECT_URI", raising=False)
|
|
monkeypatch.delenv("OMNIGENT_OIDC_COOKIE_SECRET", raising=False)
|
|
with pytest.raises(RuntimeError, match="Missing required environment variable"):
|
|
create_auth_provider()
|
|
|
|
|
|
# ── OIDCConfig.from_env validation ───────────────────────────────
|
|
|
|
|
|
def test_oidc_config_rejects_short_cookie_secret(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""Cookie secret shorter than 32 bytes is rejected at startup.
|
|
|
|
A short secret weakens the HMAC-SHA256 signing and makes
|
|
brute-force attacks feasible.
|
|
"""
|
|
monkeypatch.setenv("OMNIGENT_OIDC_ISSUER", "https://github.com")
|
|
monkeypatch.setenv("OMNIGENT_OIDC_CLIENT_ID", "test")
|
|
monkeypatch.setenv("OMNIGENT_OIDC_CLIENT_SECRET", "test")
|
|
monkeypatch.setenv("OMNIGENT_OIDC_REDIRECT_URI", "https://app/callback")
|
|
# 16 bytes = 32 hex chars, but we need 32 bytes = 64 hex chars.
|
|
monkeypatch.setenv("OMNIGENT_OIDC_COOKIE_SECRET", "aa" * 16)
|
|
with pytest.raises(RuntimeError, match="at least 32 bytes"):
|
|
OIDCConfig.from_env()
|
|
|
|
|
|
def test_oidc_config_rejects_invalid_hex_secret(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""Non-hex cookie secret is rejected at startup.
|
|
|
|
The secret must be hex-encoded so it can be safely stored in
|
|
environment variables.
|
|
"""
|
|
monkeypatch.setenv("OMNIGENT_OIDC_ISSUER", "https://github.com")
|
|
monkeypatch.setenv("OMNIGENT_OIDC_CLIENT_ID", "test")
|
|
monkeypatch.setenv("OMNIGENT_OIDC_CLIENT_SECRET", "test")
|
|
monkeypatch.setenv("OMNIGENT_OIDC_REDIRECT_URI", "https://app/callback")
|
|
monkeypatch.setenv("OMNIGENT_OIDC_COOKIE_SECRET", "not-hex-at-all")
|
|
with pytest.raises(RuntimeError, match="valid hex string"):
|
|
OIDCConfig.from_env()
|
|
|
|
|
|
def test_oidc_config_github_provider(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""GitHub issuer produces a github provider type with hardcoded endpoints.
|
|
|
|
GitHub doesn't implement OIDC discovery, so we hardcode the
|
|
endpoints. If this test fails, the GitHub login flow will break.
|
|
"""
|
|
monkeypatch.setenv("OMNIGENT_OIDC_ISSUER", "https://github.com")
|
|
monkeypatch.setenv("OMNIGENT_OIDC_CLIENT_ID", "test-client")
|
|
monkeypatch.setenv("OMNIGENT_OIDC_CLIENT_SECRET", "test-secret")
|
|
monkeypatch.setenv("OMNIGENT_OIDC_REDIRECT_URI", "https://app/callback")
|
|
monkeypatch.setenv("OMNIGENT_OIDC_COOKIE_SECRET", "aa" * 32)
|
|
|
|
config = OIDCConfig.from_env()
|
|
|
|
assert config.provider_type == "github"
|
|
assert config.authorization_endpoint == "https://github.com/login/oauth/authorize"
|
|
assert config.token_endpoint == "https://github.com/login/oauth/access_token"
|
|
assert config.jwks_uri is None, "GitHub has no JWKS URI (no id_token)."
|
|
assert "read:user" in config.scopes
|
|
assert "user:email" in config.scopes
|
|
|
|
|
|
def test_oidc_config_github_empty_scopes_env_falls_back_to_default(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""Empty OMNIGENT_OIDC_SCOPES → provider default (else GitHub 404s consent)."""
|
|
monkeypatch.setenv("OMNIGENT_OIDC_ISSUER", "https://github.com")
|
|
monkeypatch.setenv("OMNIGENT_OIDC_CLIENT_ID", "test")
|
|
monkeypatch.setenv("OMNIGENT_OIDC_CLIENT_SECRET", "test")
|
|
monkeypatch.setenv("OMNIGENT_OIDC_REDIRECT_URI", "https://app/callback")
|
|
monkeypatch.setenv("OMNIGENT_OIDC_COOKIE_SECRET", "aa" * 32)
|
|
monkeypatch.setenv("OMNIGENT_OIDC_SCOPES", "")
|
|
|
|
config = OIDCConfig.from_env()
|
|
|
|
assert "read:user" in config.scopes
|
|
assert "user:email" in config.scopes
|
|
|
|
|
|
def test_oidc_config_allowed_domains_parsing(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""OMNIGENT_OIDC_ALLOWED_DOMAINS is parsed into a frozenset.
|
|
|
|
Domains are lowercased and trimmed. If parsing is broken,
|
|
the domain allowlist check in the callback would accept or
|
|
reject the wrong domains.
|
|
"""
|
|
monkeypatch.setenv("OMNIGENT_OIDC_ISSUER", "https://github.com")
|
|
monkeypatch.setenv("OMNIGENT_OIDC_CLIENT_ID", "test")
|
|
monkeypatch.setenv("OMNIGENT_OIDC_CLIENT_SECRET", "test")
|
|
monkeypatch.setenv("OMNIGENT_OIDC_REDIRECT_URI", "https://app/callback")
|
|
monkeypatch.setenv("OMNIGENT_OIDC_COOKIE_SECRET", "aa" * 32)
|
|
monkeypatch.setenv("OMNIGENT_OIDC_ALLOWED_DOMAINS", " Example.COM , test.org ")
|
|
|
|
config = OIDCConfig.from_env()
|
|
|
|
assert config.allowed_domains == frozenset({"example.com", "test.org"})
|
|
|
|
|
|
def test_oidc_redirect_uri_derived_from_domain(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""With REDIRECT_URI unset, it's derived as https://<OMNIGENT_DOMAIN>/auth/callback.
|
|
|
|
Lets a domain-based deploy set one var (OMNIGENT_DOMAIN, already
|
|
needed by the Caddy overlay) instead of two, and removes the
|
|
http/https scheme mistake. GitHub branch → no network discovery.
|
|
"""
|
|
monkeypatch.setenv("OMNIGENT_OIDC_ISSUER", "https://github.com")
|
|
monkeypatch.setenv("OMNIGENT_OIDC_CLIENT_ID", "test")
|
|
monkeypatch.setenv("OMNIGENT_OIDC_CLIENT_SECRET", "test")
|
|
monkeypatch.delenv("OMNIGENT_OIDC_REDIRECT_URI", raising=False)
|
|
monkeypatch.setenv("OMNIGENT_DOMAIN", "omnigent.example.com")
|
|
monkeypatch.setenv("OMNIGENT_OIDC_COOKIE_SECRET", "aa" * 32)
|
|
|
|
config = OIDCConfig.from_env()
|
|
|
|
assert config.redirect_uri == "https://omnigent.example.com/auth/callback"
|
|
# Derived URI is https → secure cookies + __Host- prefix.
|
|
assert config.secure_cookies is True
|
|
|
|
|
|
def test_oidc_explicit_redirect_uri_wins_over_domain(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""An explicit REDIRECT_URI is used verbatim even when OMNIGENT_DOMAIN is set."""
|
|
monkeypatch.setenv("OMNIGENT_OIDC_ISSUER", "https://github.com")
|
|
monkeypatch.setenv("OMNIGENT_OIDC_CLIENT_ID", "test")
|
|
monkeypatch.setenv("OMNIGENT_OIDC_CLIENT_SECRET", "test")
|
|
monkeypatch.setenv("OMNIGENT_OIDC_REDIRECT_URI", "http://10.0.0.5:8000/auth/callback")
|
|
monkeypatch.setenv("OMNIGENT_DOMAIN", "omnigent.example.com")
|
|
monkeypatch.setenv("OMNIGENT_OIDC_COOKIE_SECRET", "aa" * 32)
|
|
|
|
config = OIDCConfig.from_env()
|
|
|
|
assert config.redirect_uri == "http://10.0.0.5:8000/auth/callback"
|
|
|
|
|
|
def test_oidc_redirect_uri_required_without_domain(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Fail loud when neither REDIRECT_URI nor OMNIGENT_DOMAIN is set."""
|
|
monkeypatch.setenv("OMNIGENT_OIDC_ISSUER", "https://github.com")
|
|
monkeypatch.setenv("OMNIGENT_OIDC_CLIENT_ID", "test")
|
|
monkeypatch.setenv("OMNIGENT_OIDC_CLIENT_SECRET", "test")
|
|
monkeypatch.delenv("OMNIGENT_OIDC_REDIRECT_URI", raising=False)
|
|
monkeypatch.delenv("OMNIGENT_DOMAIN", raising=False)
|
|
monkeypatch.setenv("OMNIGENT_OIDC_COOKIE_SECRET", "aa" * 32)
|
|
|
|
with pytest.raises(RuntimeError, match="OMNIGENT_OIDC_REDIRECT_URI"):
|
|
OIDCConfig.from_env()
|