feat: add organization policy enforcement

.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
+}

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

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.

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(

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

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."""