From f1d6013b547a08ae5be0e50ba2e8a9e59208382a Mon Sep 17 00:00:00 2001 From: Snehil Kishore Date: Fri, 18 Sep 2026 17:06:05 +0530 Subject: [PATCH 1/4] feat(org): add organization policy enforcement to ApiClientOptions Adds organization_policy ("allow"/"required") and organization_id allowlist options. When required, verify_access_token now rejects tokens missing org_id or carrying an org_id outside the allowlist. Passing organization_id with policy "allow" raises ConfigurationError at construction time. --- src/auth0_api_python/api_client.py | 30 ++++++++++++++++++++++++++++++ src/auth0_api_python/config.py | 11 +++++++++++ src/auth0_api_python/errors.py | 14 ++++++++++++++ 3 files changed, 55 insertions(+) diff --git a/src/auth0_api_python/api_client.py b/src/auth0_api_python/api_client.py index a87ef14..c5ba7c0 100644 --- a/src/auth0_api_python/api_client.py +++ b/src/auth0_api_python/api_client.py @@ -18,7 +18,9 @@ InvalidAuthSchemeError, InvalidDpopProofError, MissingAuthorizationError, + MissingOrganizationError, MissingRequiredArgumentError, + OrganizationNotAllowedError, VerifyAccessTokenError, ) from .types import OnBehalfOfTokenResult @@ -104,6 +106,16 @@ 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 not None and options.organization_policy != "required": + raise ConfigurationError( + "organization_id is only valid when organization_policy is 'required'" + ) + if options.cache_adapter: self._discovery_cache = options.cache_adapter self._jwks_cache = options.cache_adapter @@ -406,6 +418,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 +434,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 +576,20 @@ async def verify_access_token( if rc not in claims: raise VerifyAccessTokenError(f"Missing required claim: {rc}") + # Organization policy enforcement + org_id = claims.get("org_id") + if self.options.organization_policy == "required": + if not org_id: + raise MissingOrganizationError("Token missing required 'org_id' claim") + allowed_orgs = self.options.organization_id + if allowed_orgs is not None: + if isinstance(allowed_orgs, str): + allowed_orgs = [allowed_orgs] + if org_id not in allowed_orgs: + raise OrganizationNotAllowedError( + f"Organization '{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..68ebdad 100644 --- a/src/auth0_api_python/config.py +++ b/src/auth0_api_python/config.py @@ -32,6 +32,13 @@ 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 single value or a list). + Only valid when organization_policy is "required" - passing it with + "allow" raises ConfigurationError at construction time. """ def __init__( self, @@ -49,6 +56,8 @@ def __init__( client_id: Optional[str] = None, client_secret: Optional[str] = None, timeout: float = 10.0, + organization_policy: str = "allow", + organization_id: Optional[Union[str, list[str]]] = None, ): self.domain = domain self.domains = domains @@ -64,3 +73,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..10d4a9f 100644 --- a/src/auth0_api_python/errors.py +++ b/src/auth0_api_python/errors.py @@ -56,6 +56,20 @@ 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.""" + + def get_error_code(self) -> str: + return "missing_organization" + + +class OrganizationNotAllowedError(VerifyAccessTokenError): + """Error raised when the token's org_id claim is not in the organization_id allowlist.""" + + def get_error_code(self) -> str: + return "organization_not_allowed" + + class InvalidAuthSchemeError(BaseAuthError): """Error raised when the provided authentication scheme is unsupported.""" From 63466f38436324dc76f0583a9c4e2d73fa6954f1 Mon Sep 17 00:00:00 2001 From: Snehil Kishore Date: Fri, 18 Sep 2026 17:06:18 +0530 Subject: [PATCH 2/4] test(org): cover organization policy enforcement Adds coverage for missing org_id under "required" policy, org_id outside the allowlist, an allowed org_id succeeding, default "allow" policy not requiring org_id, and the construction-time ConfigurationError when organization_id is set without policy "required". Each test is co-located with the existing tests for the surface it exercises. --- tests/test_api_client.py | 217 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 217 insertions(+) diff --git a/tests/test_api_client.py b/tests/test_api_client.py index 9931c39..c10c89e 100644 --- a/tests/test_api_client.py +++ b/tests/test_api_client.py @@ -31,7 +31,9 @@ InvalidAuthSchemeError, InvalidDpopProofError, MissingAuthorizationError, + MissingOrganizationError, MissingRequiredArgumentError, + OrganizationNotAllowedError, VerifyAccessTokenError, ) from auth0_api_python.token_utils import ( @@ -470,6 +472,205 @@ 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() == "missing_organization" + + +@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() == "organization_not_allowed" + + +@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" + # DPOP PROOF VERIFICATION TESTS @@ -3257,6 +3458,21 @@ 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 async def test_mcd_resolve_allowed_domains_static_list(): """Test _resolve_allowed_domains with static list.""" @@ -4417,3 +4633,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/" + From cafde3eb02042fcfd0786f07bc696075d5cc3504 Mon Sep 17 00:00:00 2001 From: Snehil Kishore Date: Fri, 18 Sep 2026 19:02:46 +0530 Subject: [PATCH 3/4] docs(org): add Organization Policy guide and README section --- README.md | 20 +++++++ docs/OrganizationPolicy.md | 107 +++++++++++++++++++++++++++++++++++++ 2 files changed, 127 insertions(+) create mode 100644 docs/OrganizationPolicy.md 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..d699c28 --- /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. + +## 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() # "missing_organization" +``` + +### 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) # "Organization 'org_xyz' is not in the allowed list" + e.get_status_code() # 401 + e.get_error_code() # "organization_not_allowed" +``` + +> [!NOTE] +> `MissingOrganizationError` and `OrganizationNotAllowedError` are both subclasses of `VerifyAccessTokenError`. `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. From ae99659c58403a23319a20af1e6f4af919195240 Mon Sep 17 00:00:00 2001 From: Snehil Kishore Date: Thu, 8 Oct 2026 09:31:00 +0530 Subject: [PATCH 4/4] fix(org): validate organization_id and report org failures as invalid_token - Export MissingOrganizationError and OrganizationNotAllowedError from the package root - Keep invalid_token as the error code for organization failures - Validate organization_id at construction and normalize it once - Reject a non-string org_id claim and read it only under the required policy - Keep the org_id out of the error message and log it instead - Type organization_policy as Literal["required", "allow"] - Add tests for the single-string allowlist, no allowlist, allow policy, invalid values and the response header --- docs/OrganizationPolicy.md | 10 +- src/auth0_api_python/__init__.py | 4 + src/auth0_api_python/api_client.py | 31 ++-- src/auth0_api_python/config.py | 11 +- src/auth0_api_python/errors.py | 6 - tests/test_api_client.py | 230 ++++++++++++++++++++++++++++- 6 files changed, 260 insertions(+), 32 deletions(-) diff --git a/docs/OrganizationPolicy.md b/docs/OrganizationPolicy.md index d699c28..d0052f0 100644 --- a/docs/OrganizationPolicy.md +++ b/docs/OrganizationPolicy.md @@ -49,7 +49,7 @@ api_client = ApiClient(ApiClientOptions( )) ``` -`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. +`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 @@ -85,7 +85,7 @@ try: except MissingOrganizationError as e: print(e) # "Token missing required 'org_id' claim" e.get_status_code() # 401 - e.get_error_code() # "missing_organization" + e.get_error_code() # "invalid_token" ``` ### Organization Not Allowed @@ -98,10 +98,10 @@ from auth0_api_python import OrganizationNotAllowedError try: claims = await api_client.verify_access_token(access_token) except OrganizationNotAllowedError as e: - print(e) # "Organization 'org_xyz' is not in the allowed list" + print(e) # "Token org_id is not in the allowed list" e.get_status_code() # 401 - e.get_error_code() # "organization_not_allowed" + e.get_error_code() # "invalid_token" ``` > [!NOTE] -> `MissingOrganizationError` and `OrganizationNotAllowedError` are both subclasses of `VerifyAccessTokenError`. `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. +> `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 c5ba7c0..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 @@ -111,10 +112,23 @@ def __init__(self, options: ApiClientOptions): raise ConfigurationError( "organization_policy must be either 'required' or 'allow'" ) - if options.organization_id is not None and options.organization_policy != "required": + 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 @@ -577,18 +591,13 @@ async def verify_access_token( raise VerifyAccessTokenError(f"Missing required claim: {rc}") # Organization policy enforcement - org_id = claims.get("org_id") if self.options.organization_policy == "required": - if not org_id: + org_id = claims.get("org_id") + if not isinstance(org_id, str) or not org_id: raise MissingOrganizationError("Token missing required 'org_id' claim") - allowed_orgs = self.options.organization_id - if allowed_orgs is not None: - if isinstance(allowed_orgs, str): - allowed_orgs = [allowed_orgs] - if org_id not in allowed_orgs: - raise OrganizationNotAllowedError( - f"Organization '{org_id}' is not in the allowed list" - ) + 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 diff --git a/src/auth0_api_python/config.py b/src/auth0_api_python/config.py index 68ebdad..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 @@ -36,9 +36,10 @@ class ApiClientOptions: "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 single value or a list). - Only valid when organization_policy is "required" - passing it with - "allow" raises ConfigurationError at construction time. + 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, @@ -56,7 +57,7 @@ def __init__( client_id: Optional[str] = None, client_secret: Optional[str] = None, timeout: float = 10.0, - organization_policy: str = "allow", + organization_policy: Literal["required", "allow"] = "allow", organization_id: Optional[Union[str, list[str]]] = None, ): self.domain = domain diff --git a/src/auth0_api_python/errors.py b/src/auth0_api_python/errors.py index 10d4a9f..e88372c 100644 --- a/src/auth0_api_python/errors.py +++ b/src/auth0_api_python/errors.py @@ -59,16 +59,10 @@ def get_error_code(self) -> str: class MissingOrganizationError(VerifyAccessTokenError): """Error raised when organization_policy is 'required' but the token has no org_id claim.""" - def get_error_code(self) -> str: - return "missing_organization" - class OrganizationNotAllowedError(VerifyAccessTokenError): """Error raised when the token's org_id claim is not in the organization_id allowlist.""" - def get_error_code(self) -> str: - return "organization_not_allowed" - 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 c10c89e..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 ( @@ -31,9 +37,7 @@ InvalidAuthSchemeError, InvalidDpopProofError, MissingAuthorizationError, - MissingOrganizationError, MissingRequiredArgumentError, - OrganizationNotAllowedError, VerifyAccessTokenError, ) from auth0_api_python.token_utils import ( @@ -520,7 +524,8 @@ async def test_organization_policy_missing_org_id_when_required(httpx_mock: HTTP with pytest.raises(MissingOrganizationError) as err: await api_client.verify_access_token(access_token=access_token) - assert err.value.get_error_code() == "missing_organization" + assert err.value.get_error_code() == "invalid_token" + assert err.value.get_status_code() == 401 @pytest.mark.asyncio @@ -571,7 +576,9 @@ async def test_organization_policy_disallowed_org(httpx_mock: HTTPXMock): with pytest.raises(OrganizationNotAllowedError) as err: await api_client.verify_access_token(access_token=access_token) - assert err.value.get_error_code() == "organization_not_allowed" + 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 @@ -672,6 +679,154 @@ async def test_organization_policy_default_allow_does_not_require_org_id(httpx_m 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 # --- Core Success Tests --- @@ -1963,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( @@ -3473,6 +3665,34 @@ async def test_organization_id_with_allow_policy_raises_at_construction(httpx_mo 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."""