diff --git a/.ruff.toml b/.ruff.toml index b500d05..06e5119 100644 --- a/.ruff.toml +++ b/.ruff.toml @@ -1,5 +1,7 @@ line-length = 100 target-version = "py39" + +[lint] select = [ "E", # pycodestyle errors "W", # pycodestyle warnings @@ -12,5 +14,6 @@ select = [ ] ignore = ["E501", "B904"] # Line too long (handled by black), Exception handling without from -[per-file-ignores] -"tests/*" = ["S101", "S105", "S106"] # Allow assert and ignore hardcoded password warnings in test files +per-file-ignores = { + "tests/*" = ["S101", "S105", "S106"], # Allow assert and ignore hardcoded password warnings in test files +} 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. 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.""" 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/" +