diff --git a/README.md b/README.md index 48df61f..932ddd2 100644 --- a/README.md +++ b/README.md @@ -18,6 +18,7 @@ This SDK provides comprehensive support for securing APIs with Auth0-issued acce ### **Core Features** - **Unified Entry Point**: `verify_request()` - automatically detects and validates Bearer or DPoP schemes - **Multi-Custom Domain (MCD)** - Accept tokens from multiple Auth0 domains with static lists or dynamic resolvers +- **Organization Policy** - Enforce and optionally allowlist the `org_id` claim on incoming tokens - **OIDC Discovery** - Automatic fetching of Auth0 metadata and JWKS with per-issuer caching - **JWT Validation** - Complete RS256 signature verification with claim validation - **DPoP Proof Verification** - Full RFC 9449 compliance with ES256 signature validation @@ -413,6 +414,25 @@ For hybrid mode (migration scenarios), resolver patterns, error handling, and ca An anonymous token passes verification by default. Deciding whether an anonymous caller is authorized is your application's responsibility. See [Anonymous Callers](EXAMPLES.md#anonymous-callers) for allow, block-per-route, and block-globally patterns. +### 9. Organization Policy + +For APIs that need to enforce an Auth0 Organization context on every request, optionally restricted to a specific set of Organizations: + +```python +from auth0_api_python import ApiClient, ApiClientOptions + +api_client = ApiClient(ApiClientOptions( + domain="tenant.auth0.com", + audience="https://api.example.com", + organization_policy="required", + organization_id=["org_abc123", "org_def456"] +)) + +claims = await api_client.verify_access_token(access_token) +``` + +See the **[Organization Policy Guide](docs/OrganizationPolicy.md)** for policy modes, the allowlist, and error handling. + ## Feedback ### Contributing diff --git a/docs/OrganizationPolicy.md b/docs/OrganizationPolicy.md new file mode 100644 index 0000000..d0052f0 --- /dev/null +++ b/docs/OrganizationPolicy.md @@ -0,0 +1,107 @@ +# Organization Policy + +The SDK can enforce that incoming access tokens carry a valid Auth0 Organization (`org_id` claim). This is useful for APIs that serve B2B customers where every request must be scoped to an Organization, optionally restricted to a specific set of Organizations. + +## Policy Modes + +### Allow (Default) + +By default, `organization_policy` is `"allow"`. The SDK uses the `org_id` claim when present but does not require it. This is the pre-existing behavior of `verify_access_token`, so existing callers see no change unless they opt in to `"required"`. + +```python +from auth0_api_python import ApiClient, ApiClientOptions + +api_client = ApiClient(ApiClientOptions( + domain="tenant.auth0.com", + audience="https://api.example.com" + # organization_policy defaults to "allow" +)) + +# Tokens with or without an org_id claim are both accepted +claims = await api_client.verify_access_token(access_token) +``` + +### Required + +Set `organization_policy="required"` to reject any token that has no `org_id` claim: + +```python +api_client = ApiClient(ApiClientOptions( + domain="tenant.auth0.com", + audience="https://api.example.com", + organization_policy="required" +)) + +# Raises MissingOrganizationError if the token has no org_id claim +claims = await api_client.verify_access_token(access_token) +``` + +## Organization Allowlist + +When `organization_policy="required"`, you can additionally restrict which Organizations are accepted with `organization_id`. It takes a single `org_id` claim value or a list of them: + +```python +api_client = ApiClient(ApiClientOptions( + domain="tenant.auth0.com", + audience="https://api.example.com", + organization_policy="required", + organization_id=["org_abc123", "org_def456"] +)) +``` + +`organization_id` compares the opaque `org_id` claim value directly (string comparison, no network call). It does not accept or resolve the human-readable Organization name. It must be a non-empty string or a non-empty list of non-empty strings, otherwise `ConfigurationError` is raised at construction. + +## Error Handling + +### Configuration Errors + +Raised at initialization when the SDK configuration is invalid: + +```python +from auth0_api_python import ApiClient, ApiClientOptions, ConfigurationError + +# organization_id passed with the default "allow" policy +try: + api_client = ApiClient(ApiClientOptions( + domain="tenant.auth0.com", + audience="https://api.example.com", + organization_id="org_abc123" + )) +except ConfigurationError as e: + print(e) # "organization_id is only valid when organization_policy is 'required'" + e.get_status_code() # 500 + e.get_error_code() # "invalid_configuration" +``` + +### Missing Organization + +Raised when `organization_policy="required"` and the token has no `org_id` claim: + +```python +from auth0_api_python import MissingOrganizationError + +try: + claims = await api_client.verify_access_token(access_token) +except MissingOrganizationError as e: + print(e) # "Token missing required 'org_id' claim" + e.get_status_code() # 401 + e.get_error_code() # "invalid_token" +``` + +### Organization Not Allowed + +Raised when the token's `org_id` is not in the `organization_id` allowlist: + +```python +from auth0_api_python import OrganizationNotAllowedError + +try: + claims = await api_client.verify_access_token(access_token) +except OrganizationNotAllowedError as e: + print(e) # "Token org_id is not in the allowed list" + e.get_status_code() # 401 + e.get_error_code() # "invalid_token" +``` + +> [!NOTE] +> `MissingOrganizationError` and `OrganizationNotAllowedError` are both subclasses of `VerifyAccessTokenError`, so they carry the same `invalid_token` error code and 401 status. The rejected `org_id` is not included in the error message or response headers. It is logged as a warning instead. `WWW-Authenticate` response headers (via `get_headers()`) are only populated when the token is verified through `verify_request()`, which wraps these errors before re-raising. Calling `verify_access_token()` directly does not attach response headers. diff --git a/src/auth0_api_python/__init__.py b/src/auth0_api_python/__init__.py index cc77b05..aadb9c6 100644 --- a/src/auth0_api_python/__init__.py +++ b/src/auth0_api_python/__init__.py @@ -14,6 +14,8 @@ ConfigurationError, DomainsResolverError, GetTokenByExchangeProfileError, + MissingOrganizationError, + OrganizationNotAllowedError, ) from .types import ( DomainsResolver, @@ -34,5 +36,7 @@ "get_current_actor", "get_delegation_chain", "InMemoryCache", + "MissingOrganizationError", "OnBehalfOfTokenResult", + "OrganizationNotAllowedError", ] diff --git a/src/auth0_api_python/api_client.py b/src/auth0_api_python/api_client.py index a87ef14..9e7409a 100644 --- a/src/auth0_api_python/api_client.py +++ b/src/auth0_api_python/api_client.py @@ -1,4 +1,5 @@ import asyncio +import logging import time from collections.abc import Mapping, Sequence from typing import Any, Optional, Union @@ -18,7 +19,9 @@ InvalidAuthSchemeError, InvalidDpopProofError, MissingAuthorizationError, + MissingOrganizationError, MissingRequiredArgumentError, + OrganizationNotAllowedError, VerifyAccessTokenError, ) from .types import OnBehalfOfTokenResult @@ -104,6 +107,29 @@ def __init__(self, options: ApiClientOptions): if not isinstance(options.cache_max_entries, int) or options.cache_max_entries < 2: raise ConfigurationError("cache_max_entries must be an integer greater than 1") + # Validate organization policy configuration + if options.organization_policy not in ("required", "allow"): + raise ConfigurationError( + "organization_policy must be either 'required' or 'allow'" + ) + if options.organization_id is None: + self._allowed_org_ids = None + elif options.organization_policy != "required": + raise ConfigurationError( + "organization_id is only valid when organization_policy is 'required'" + ) + else: + org_ids = options.organization_id + if isinstance(org_ids, str): + org_ids = [org_ids] + if not isinstance(org_ids, list) or not org_ids or not all( + isinstance(o, str) and o.strip() for o in org_ids + ): + raise ConfigurationError( + "organization_id must be a non-empty string or a non-empty list of non-empty strings" + ) + self._allowed_org_ids = frozenset(org_ids) + if options.cache_adapter: self._discovery_cache = options.cache_adapter self._jwks_cache = options.cache_adapter @@ -406,6 +432,8 @@ async def verify_access_token( - Decodes and validates signature (RS256) with the correct key. - Checks standard claims: 'iss', 'aud', 'exp', 'iat' - Checks extra required claims if 'required_claims' is provided. + - Enforces organization_policy: requires 'org_id' when set to "required", + and checks it against organization_id when an allowlist is configured. Args: access_token: The JWT access token to verify @@ -420,6 +448,8 @@ async def verify_access_token( MissingRequiredArgumentError: If no token is provided. VerifyAccessTokenError: If verification fails (signature, claims mismatch, etc.). DomainsResolverError: If domains resolver function fails. + MissingOrganizationError: If organization_policy is "required" and the token has no org_id claim. + OrganizationNotAllowedError: If the token's org_id is not in the organization_id allowlist. """ if not access_token: raise MissingRequiredArgumentError("access_token") @@ -560,6 +590,15 @@ async def verify_access_token( if rc not in claims: raise VerifyAccessTokenError(f"Missing required claim: {rc}") + # Organization policy enforcement + if self.options.organization_policy == "required": + org_id = claims.get("org_id") + if not isinstance(org_id, str) or not org_id: + raise MissingOrganizationError("Token missing required 'org_id' claim") + if self._allowed_org_ids is not None and org_id not in self._allowed_org_ids: + logging.warning("Rejected token with org_id %r not in the organization_id allowlist", org_id) + raise OrganizationNotAllowedError("Token org_id is not in the allowed list") + return claims async def verify_dpop_proof( diff --git a/src/auth0_api_python/config.py b/src/auth0_api_python/config.py index 1929d55..6fb7fda 100644 --- a/src/auth0_api_python/config.py +++ b/src/auth0_api_python/config.py @@ -2,7 +2,7 @@ Configuration classes and utilities for auth0-api-python. """ -from typing import TYPE_CHECKING, Callable, Optional, Union +from typing import TYPE_CHECKING, Callable, Literal, Optional, Union if TYPE_CHECKING: from .cache import CacheAdapter @@ -32,6 +32,14 @@ class ApiClientOptions: client_secret: Required for get_access_token_for_connection, get_token_by_exchange_profile, and get_token_on_behalf_of. timeout: HTTP timeout in seconds for token endpoint requests (default: 10.0). + organization_policy: Whether the incoming token must carry an org_id claim. + "allow" (default) uses org_id when present but does not require it, + matching the pre-existing behavior of verify_access_token. + "required" rejects any token without an org_id claim. + organization_id: Optional allowlist of org_id claim values (a non-empty string or a non-empty + list of non-empty strings). Only valid when organization_policy is + "required" - passing it with "allow" raises ConfigurationError at + construction time, as does an empty or non-string value. """ def __init__( self, @@ -49,6 +57,8 @@ def __init__( client_id: Optional[str] = None, client_secret: Optional[str] = None, timeout: float = 10.0, + organization_policy: Literal["required", "allow"] = "allow", + organization_id: Optional[Union[str, list[str]]] = None, ): self.domain = domain self.domains = domains @@ -64,3 +74,5 @@ def __init__( self.client_id = client_id self.client_secret = client_secret self.timeout = timeout + self.organization_policy = organization_policy + self.organization_id = organization_id diff --git a/src/auth0_api_python/errors.py b/src/auth0_api_python/errors.py index a0a43d0..e88372c 100644 --- a/src/auth0_api_python/errors.py +++ b/src/auth0_api_python/errors.py @@ -56,6 +56,14 @@ def get_error_code(self) -> str: return "invalid_token" +class MissingOrganizationError(VerifyAccessTokenError): + """Error raised when organization_policy is 'required' but the token has no org_id claim.""" + + +class OrganizationNotAllowedError(VerifyAccessTokenError): + """Error raised when the token's org_id claim is not in the organization_id allowlist.""" + + class InvalidAuthSchemeError(BaseAuthError): """Error raised when the provided authentication scheme is unsupported.""" diff --git a/tests/test_api_client.py b/tests/test_api_client.py index 9931c39..0a1beda 100644 --- a/tests/test_api_client.py +++ b/tests/test_api_client.py @@ -19,7 +19,13 @@ from freezegun import freeze_time from pytest_httpx import HTTPXMock -from auth0_api_python import get_current_actor, get_delegation_chain +from auth0_api_python import ( + MissingOrganizationError, + OrganizationNotAllowedError, + get_current_actor, + get_delegation_chain, +) +from auth0_api_python import __all__ as package_exports from auth0_api_python.api_client import MAX_ARRAY_VALUES_PER_KEY, ApiClient from auth0_api_python.config import ApiClientOptions from auth0_api_python.errors import ( @@ -470,6 +476,356 @@ async def test_verify_access_token_fail_malformed_token(): assert "failed to parse token" in str(e.value).lower() +# ===== Organization Policy: verify_access_token Enforcement ===== + +@pytest.mark.asyncio +async def test_organization_policy_missing_org_id_when_required(httpx_mock: HTTPXMock): + """Test that a token with no org_id claim is rejected when organization_policy is 'required'.""" + httpx_mock.add_response( + method="GET", + url=DISCOVERY_URL, + json={ + "issuer": "https://auth0.local/", + "jwks_uri": JWKS_URL + } + ) + httpx_mock.add_response( + method="GET", + url=JWKS_URL, + json={ + "keys": [ + { + "kty": "RSA", + "kid": "TEST_KEY", + "n": "whYOFK2Ocbbpb_zVypi9SeKiNUqKQH0zTKN1-6fpCTu6ZalGI82s7XK3tan4dJt90ptUPKD2zvxqTzFNfx4HHHsrYCf2-FMLn1VTJfQazA2BvJqAwcpW1bqRUEty8tS_Yv4hRvWfQPcc2Gc3-_fQOOW57zVy-rNoJc744kb30NjQxdGp03J2S3GLQu7oKtSDDPooQHD38PEMNnITf0pj-KgDPjymkMGoJlO3aKppsjfbt_AH6GGdRghYRLOUwQU-h-ofWHR3lbYiKtXPn5dN24kiHy61e3VAQ9_YAZlwXC_99GGtw_NpghFAuM4P1JDn0DppJldy3PGFC0GfBCZASw", + "e": "AQAB", + "alg": "RS256", + "use": "sig" + } + ] + } + ) + + access_token = await generate_token( + domain="auth0.local", + user_id="user_123", + audience="my-audience", + issuer=None, + iat=True, + exp=True, + ) + + api_client = ApiClient(ApiClientOptions( + domain="auth0.local", + audience="my-audience", + organization_policy="required", + )) + + with pytest.raises(MissingOrganizationError) as err: + await api_client.verify_access_token(access_token=access_token) + + assert err.value.get_error_code() == "invalid_token" + assert err.value.get_status_code() == 401 + + +@pytest.mark.asyncio +async def test_organization_policy_disallowed_org(httpx_mock: HTTPXMock): + """Test that a token whose org_id is not in the allowlist is rejected.""" + httpx_mock.add_response( + method="GET", + url=DISCOVERY_URL, + json={ + "issuer": "https://auth0.local/", + "jwks_uri": JWKS_URL + } + ) + httpx_mock.add_response( + method="GET", + url=JWKS_URL, + json={ + "keys": [ + { + "kty": "RSA", + "kid": "TEST_KEY", + "n": "whYOFK2Ocbbpb_zVypi9SeKiNUqKQH0zTKN1-6fpCTu6ZalGI82s7XK3tan4dJt90ptUPKD2zvxqTzFNfx4HHHsrYCf2-FMLn1VTJfQazA2BvJqAwcpW1bqRUEty8tS_Yv4hRvWfQPcc2Gc3-_fQOOW57zVy-rNoJc744kb30NjQxdGp03J2S3GLQu7oKtSDDPooQHD38PEMNnITf0pj-KgDPjymkMGoJlO3aKppsjfbt_AH6GGdRghYRLOUwQU-h-ofWHR3lbYiKtXPn5dN24kiHy61e3VAQ9_YAZlwXC_99GGtw_NpghFAuM4P1JDn0DppJldy3PGFC0GfBCZASw", + "e": "AQAB", + "alg": "RS256", + "use": "sig" + } + ] + } + ) + + access_token = await generate_token( + domain="auth0.local", + user_id="user_123", + audience="my-audience", + issuer=None, + iat=True, + exp=True, + claims={"org_id": "org_untrusted"}, + ) + + api_client = ApiClient(ApiClientOptions( + domain="auth0.local", + audience="my-audience", + organization_policy="required", + organization_id=["org_abc123", "org_def456"], + )) + + with pytest.raises(OrganizationNotAllowedError) as err: + await api_client.verify_access_token(access_token=access_token) + + assert err.value.get_error_code() == "invalid_token" + assert err.value.get_status_code() == 401 + assert "org_untrusted" not in str(err.value) + + +@pytest.mark.asyncio +async def test_organization_policy_allowed_org_succeeds(httpx_mock: HTTPXMock): + """Test that a token with an allowlisted org_id verifies successfully.""" + httpx_mock.add_response( + method="GET", + url=DISCOVERY_URL, + json={ + "issuer": "https://auth0.local/", + "jwks_uri": JWKS_URL + } + ) + httpx_mock.add_response( + method="GET", + url=JWKS_URL, + json={ + "keys": [ + { + "kty": "RSA", + "kid": "TEST_KEY", + "n": "whYOFK2Ocbbpb_zVypi9SeKiNUqKQH0zTKN1-6fpCTu6ZalGI82s7XK3tan4dJt90ptUPKD2zvxqTzFNfx4HHHsrYCf2-FMLn1VTJfQazA2BvJqAwcpW1bqRUEty8tS_Yv4hRvWfQPcc2Gc3-_fQOOW57zVy-rNoJc744kb30NjQxdGp03J2S3GLQu7oKtSDDPooQHD38PEMNnITf0pj-KgDPjymkMGoJlO3aKppsjfbt_AH6GGdRghYRLOUwQU-h-ofWHR3lbYiKtXPn5dN24kiHy61e3VAQ9_YAZlwXC_99GGtw_NpghFAuM4P1JDn0DppJldy3PGFC0GfBCZASw", + "e": "AQAB", + "alg": "RS256", + "use": "sig" + } + ] + } + ) + + access_token = await generate_token( + domain="auth0.local", + user_id="user_123", + audience="my-audience", + issuer=None, + iat=True, + exp=True, + claims={"org_id": "org_abc123"}, + ) + + api_client = ApiClient(ApiClientOptions( + domain="auth0.local", + audience="my-audience", + organization_policy="required", + organization_id=["org_abc123", "org_def456"], + )) + + claims = await api_client.verify_access_token(access_token=access_token) + + assert claims["sub"] == "user_123" + assert claims["org_id"] == "org_abc123" + + +@pytest.mark.asyncio +async def test_organization_policy_default_allow_does_not_require_org_id(httpx_mock: HTTPXMock): + """Test that the default 'allow' policy accepts a token with no org_id claim (no behavior change).""" + httpx_mock.add_response( + method="GET", + url=DISCOVERY_URL, + json={ + "issuer": "https://auth0.local/", + "jwks_uri": JWKS_URL + } + ) + httpx_mock.add_response( + method="GET", + url=JWKS_URL, + json={ + "keys": [ + { + "kty": "RSA", + "kid": "TEST_KEY", + "n": "whYOFK2Ocbbpb_zVypi9SeKiNUqKQH0zTKN1-6fpCTu6ZalGI82s7XK3tan4dJt90ptUPKD2zvxqTzFNfx4HHHsrYCf2-FMLn1VTJfQazA2BvJqAwcpW1bqRUEty8tS_Yv4hRvWfQPcc2Gc3-_fQOOW57zVy-rNoJc744kb30NjQxdGp03J2S3GLQu7oKtSDDPooQHD38PEMNnITf0pj-KgDPjymkMGoJlO3aKppsjfbt_AH6GGdRghYRLOUwQU-h-ofWHR3lbYiKtXPn5dN24kiHy61e3VAQ9_YAZlwXC_99GGtw_NpghFAuM4P1JDn0DppJldy3PGFC0GfBCZASw", + "e": "AQAB", + "alg": "RS256", + "use": "sig" + } + ] + } + ) + + access_token = await generate_token( + domain="auth0.local", + user_id="user_123", + audience="my-audience", + issuer=None, + iat=True, + exp=True, + ) + + api_client = ApiClient(ApiClientOptions( + domain="auth0.local", + audience="my-audience", + )) + + claims = await api_client.verify_access_token(access_token=access_token) + + assert claims["sub"] == "user_123" + + +@pytest.mark.asyncio +async def test_organization_policy_single_string_allowlist(httpx_mock: HTTPXMock): + """Test that a single string organization_id accepts a matching org_id and rejects others.""" + httpx_mock.add_response( + method="GET", + url=DISCOVERY_URL, + json={"jwks_uri": JWKS_URL, "issuer": "https://auth0.local/"}, + ) + httpx_mock.add_response( + method="GET", + url=JWKS_URL, + json={"keys": [PUBLIC_RSA_JWK]}, + ) + + api_client = ApiClient(ApiClientOptions( + domain="auth0.local", + audience="my-audience", + organization_policy="required", + organization_id="org_abc123", + )) + + allowed = await generate_token( + domain="auth0.local", user_id="user_123", audience="my-audience", claims={"org_id": "org_abc123"}, + ) + claims = await api_client.verify_access_token(access_token=allowed) + assert claims["org_id"] == "org_abc123" + + other = await generate_token( + domain="auth0.local", user_id="user_123", audience="my-audience", claims={"org_id": "org_other"}, + ) + with pytest.raises(OrganizationNotAllowedError): + await api_client.verify_access_token(access_token=other) + + +@pytest.mark.asyncio +async def test_organization_policy_required_without_allowlist_accepts_any_org(httpx_mock: HTTPXMock): + """Test that 'required' with no allowlist accepts any token that carries an org_id.""" + httpx_mock.add_response( + method="GET", + url=DISCOVERY_URL, + json={"jwks_uri": JWKS_URL, "issuer": "https://auth0.local/"}, + ) + httpx_mock.add_response( + method="GET", + url=JWKS_URL, + json={"keys": [PUBLIC_RSA_JWK]}, + ) + + token = await generate_token( + domain="auth0.local", user_id="user_123", audience="my-audience", claims={"org_id": "org_any"}, + ) + api_client = ApiClient(ApiClientOptions( + domain="auth0.local", audience="my-audience", organization_policy="required", + )) + + claims = await api_client.verify_access_token(access_token=token) + + assert claims["org_id"] == "org_any" + + +@pytest.mark.asyncio +async def test_organization_policy_allow_accepts_token_with_org_id(httpx_mock: HTTPXMock): + """Test that the default 'allow' policy accepts a token that carries an org_id.""" + httpx_mock.add_response( + method="GET", + url=DISCOVERY_URL, + json={"jwks_uri": JWKS_URL, "issuer": "https://auth0.local/"}, + ) + httpx_mock.add_response( + method="GET", + url=JWKS_URL, + json={"keys": [PUBLIC_RSA_JWK]}, + ) + + token = await generate_token( + domain="auth0.local", user_id="user_123", audience="my-audience", claims={"org_id": "org_any"}, + ) + api_client = ApiClient(ApiClientOptions(domain="auth0.local", audience="my-audience")) + + claims = await api_client.verify_access_token(access_token=token) + + assert claims["org_id"] == "org_any" + + +@pytest.mark.asyncio +@pytest.mark.parametrize("org_id", [123, ["org_abc123"], "", None]) +async def test_organization_policy_rejects_non_string_org_id(httpx_mock: HTTPXMock, org_id): + """Test that an org_id claim that is not a non-empty string is treated as missing.""" + httpx_mock.add_response( + method="GET", + url=DISCOVERY_URL, + json={"jwks_uri": JWKS_URL, "issuer": "https://auth0.local/"}, + ) + httpx_mock.add_response( + method="GET", + url=JWKS_URL, + json={"keys": [PUBLIC_RSA_JWK]}, + ) + + token = await generate_token( + domain="auth0.local", user_id="user_123", audience="my-audience", claims={"org_id": org_id}, + ) + api_client = ApiClient(ApiClientOptions( + domain="auth0.local", + audience="my-audience", + organization_policy="required", + organization_id=["org_abc123"], + )) + + with pytest.raises(MissingOrganizationError): + await api_client.verify_access_token(access_token=token) + + +@pytest.mark.asyncio +async def test_organization_policy_logs_rejected_org_id(httpx_mock: HTTPXMock, caplog): + """Test that the rejected org_id is logged since it is kept out of the error message.""" + httpx_mock.add_response( + method="GET", + url=DISCOVERY_URL, + json={"jwks_uri": JWKS_URL, "issuer": "https://auth0.local/"}, + ) + httpx_mock.add_response( + method="GET", + url=JWKS_URL, + json={"keys": [PUBLIC_RSA_JWK]}, + ) + + token = await generate_token( + domain="auth0.local", user_id="user_123", audience="my-audience", claims={"org_id": "org_untrusted"}, + ) + api_client = ApiClient(ApiClientOptions( + domain="auth0.local", + audience="my-audience", + organization_policy="required", + organization_id=["org_abc123"], + )) + + with caplog.at_level("WARNING"), pytest.raises(OrganizationNotAllowedError): + await api_client.verify_access_token(access_token=token) + + assert "org_untrusted" in caplog.text + + +def test_organization_errors_are_exported_from_package_root(): + """Test that the organization errors are listed in the package-root exports.""" + assert {"MissingOrganizationError", "OrganizationNotAllowedError"} <= set(package_exports) + # DPOP PROOF VERIFICATION TESTS @@ -1762,6 +2118,43 @@ async def test_verify_request_dpop_required_mismatch(dpop_required, auth_header, assert err.value.get_status_code() == 400 assert "invalid_request" in str(err.value.get_error_code()).lower() + +@pytest.mark.asyncio +async def test_organization_policy_verify_request_header_uses_invalid_token_and_static_text(httpx_mock: HTTPXMock): + """Test that org failures surface as invalid_token in WWW-Authenticate without echoing the org_id.""" + httpx_mock.add_response( + method="GET", + url=DISCOVERY_URL, + json={"jwks_uri": JWKS_URL, "issuer": "https://auth0.local/"}, + ) + httpx_mock.add_response( + method="GET", + url=JWKS_URL, + json={"keys": [PUBLIC_RSA_JWK]}, + ) + + token = await generate_token( + domain="auth0.local", user_id="user_123", audience="my-audience", claims={"org_id": "org_untrusted"}, + ) + api_client = ApiClient(ApiClientOptions( + domain="auth0.local", + audience="my-audience", + organization_policy="required", + organization_id=["org_abc123"], + )) + + with pytest.raises(OrganizationNotAllowedError) as err: + await api_client.verify_request( + headers={"authorization": f"Bearer {token}"}, + http_method="GET", + http_url="https://api.example.com/resource", + ) + + header = err.value.get_headers()["WWW-Authenticate"] + assert 'error="invalid_token"' in header + assert "org_untrusted" not in header + + @pytest.mark.asyncio async def test_get_access_token_for_connection_success(httpx_mock: HTTPXMock): httpx_mock.add_response( @@ -3257,6 +3650,49 @@ async def test_cache_config_validation(): assert client._cache_ttl == 0 +@pytest.mark.asyncio +async def test_organization_id_with_allow_policy_raises_at_construction(httpx_mock: HTTPXMock): + """Test that organization_id with organization_policy='allow' raises ConfigurationError + at ApiClient construction time, before any request is made.""" + with pytest.raises(ConfigurationError, match="organization_id is only valid when organization_policy is 'required'"): + ApiClient(ApiClientOptions( + domain="auth0.local", + audience="my-audience", + organization_policy="allow", + organization_id="org_abc123", + )) + + assert_no_requests(httpx_mock) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("organization_id", [[], "", 123, [" "], [""], ["org_abc123", 5], ("org_abc123",)]) +async def test_organization_id_invalid_value_raises_at_construction(httpx_mock: HTTPXMock, organization_id): + """Test that an empty or non-string organization_id raises ConfigurationError at construction.""" + with pytest.raises(ConfigurationError, match="organization_id must be a non-empty string or a non-empty list"): + ApiClient(ApiClientOptions( + domain="auth0.local", + audience="my-audience", + organization_policy="required", + organization_id=organization_id, + )) + + assert_no_requests(httpx_mock) + + +@pytest.mark.asyncio +async def test_invalid_organization_policy_raises_at_construction(httpx_mock: HTTPXMock): + """Test that an organization_policy other than 'required' or 'allow' raises ConfigurationError.""" + with pytest.raises(ConfigurationError, match="organization_policy must be either 'required' or 'allow'"): + ApiClient(ApiClientOptions( + domain="auth0.local", + audience="my-audience", + organization_policy="none", + )) + + assert_no_requests(httpx_mock) + + @pytest.mark.asyncio async def test_mcd_resolve_allowed_domains_static_list(): """Test _resolve_allowed_domains with static list.""" @@ -4417,3 +4853,4 @@ def capturing_resolver(context): assert ctx["request_headers"]["authorization"] == f"Bearer {token}" assert ctx["request_headers"]["x-custom-header"] == "test-value" assert ctx["unverified_iss"] == "https://tenant1.auth0.com/" +