Multi-Tenant Support for SCIM2 Server

[Th3R3p0]

The current implementation lacks proper multi-tenant isolation, which is needed when serving multiple identity providers through a single API endpoint. The server currently uses a single bearer token for authentication without any mechanism to segregate data between different tenants.

We propose enhancing the implementation with tenant isolation by:

1. Extending the authentication system to map bearer tokens to specific tenant IDs
2. Modifying the backend interface to include tenant ID in all resource operations, ensuring data isolation regardless of the storage mechanism used
3. Updating the resource filtering to respect tenant boundaries.

This approach maintains compatibility with the SCIM protocol while adding the necessary security boundaries between tenants, making the server suitable for multi-tenant environments where different organizations connect their identity providers to the same SCIM endpoint.

Is this something you would be interested in merging into the code? I can submit a pull request, if so.

[hannes-danzl-eb]

You can check out my fork which adds multi-tenant support and the possibility to deploy it as an AWS Lambda / Gateway. Note that this is part of a POC at the moment and as such still needs some work in regards to code structure and tests, but should be functionally sound.

[vpatov]

@hannesdanzl I see that you've modified the provider implementation, and backend interface, and added a tenant_id to the necessary methods. I think there is a simpler approach that doesn't change the library interface.

You can use context vars to set values for context variables that will persist throughout the processing of a request:

"""Provides helper methods for managing the auth context to be used in SCIM API calls.

Similar to the GQL API's strawberry.Info object, this pattern allows all parts of SCIM
API processing to have access to the auth context. contextvars.ContextVar guarantees
that each thread (or async task if using asynccontextmanager) has its own context variable, which means that processing
concurrent tasks is safe (would be unnaceptable otherwise).

"""
from contextlib import contextmanager
from contextvars import ContextVar

from my_app.server.middleware.auth_context import TenantInfo

_request_tenant_info: ContextVar[TenantInfo] = ContextVar(
    "request_tenant_info"
)


@contextmanager
def set_tenant_info(tenant_info):
    token = _request_tenant_info.set(tenant_info)
    try:
        yield
    finally:
        _request_tenant_info.reset(token)


def get_tenant_info():
    return _request_tenant_info.get()

In the provider dispatch override, add a function call that fetches your tenant_id, and sets the context with that tenant_id:

        tenant_info = None
        if "ServiceProviderConfig" not in request.url.path:
            try:
                tenant_info = self.authenticate_and_get_tenant_info(request)
            except HTTPException as exc:
                logger.debug(exc)
                return make_error(Error(status=exc.status_code, detail=exc.detail))

        # open context for scope of request
        with self.backend, set_tenant_info(tenant_info):
            response = await call_next(request)

Now, inside of your backend implementation, you can do:

from my_app.core.scim.context import get_tenant_info

...

    def query_resources(
        self,
        search_request: scim2_models.SearchRequest,
        resource_type_id: str | None = None,
    ) -> tuple[int, list[scim2_models.Resource]]:
        """Override query_resources for user queries using the tenant_id from tenant_info
        context.
        """
        tenant_info = get_tenant_info()

        if resource_type_id == USER_SCIM_RESOURCE_TYPE:
            tenant_id = tenant_info.id
            # data layer methods require tenant_id for all queries
            users = await scim_orm.query_users(search_request, tenant_id, resource_type_id)
            return len(users), users
        else:
            # omitting groups to make example shorter
            raise NotImplementedError()

---

The current implementation lacks proper multi-tenant isolation, which is needed when serving multiple identity providers through a single API endpoint. The server currently uses a single bearer token for authentication without any mechanism to segregate data between different tenants.

The library authors recommend using these modules as a starting point for implementing SCIM functionality, rather than a strict import + subclass (my interpretation). You should definitely override the check_auth function, and get rid of the static bearer tokens if you are doing traditional multi-tenant SaaS style auth.

[hannes-danzl-eb]

Thanks a lot Vasia, this is excellent information and just at the right time as I've started to work on a multi-tenancy real-world POC. I'll explore and if (I don't doubt) it works I'll push it.

[vpatov]

No problem. One thing to keep in mind @hannesdanzl when trying to make the code above work, is that I originally wrote it using async (I have ported the scim2_server.provider to an async implementation), and using "auth_context" rather than "tenant_info", and then quickly modified it in the markdown editor to sync before sharing to make it useful for you, but I haven't executed that code exactly as-is. What I have is basically identical except for the minor modifications I've described. I'm using it and it works well.

[j-klawson]

I'm have a need for this feature as well and would like to send the patch back. @azmeuk is @vpatov's approach acceptable to be included in the project or is that beyond the functionality you want to publish in the public repo?

[azmeuk]

Hi. Thank you all for the discussion.
I need a bit of time to study the implications.
@ccoors, do you have an advice?

[ccoors]

We're using code based on this in production with multi-tenancy. RFC 7644 recommends various approaches in Section 6.1:

• A URL prefix: "https://www.example.com/Tenants/{tenant_id}/v2/Users".
• A sub-domain: "https://{tenant_id}.example.com/v2/Groups".
• An HTTP header: The service provider may recognize a {tenant_id} provided by the client in an HTTP header as the indicator of the desired target Tenant.

We went with the first option, having the tenant_id between the prefix and /v2/.... Doing it implicitly via the authentication is technically an option, but I can imagine scenarios in which you'd want to do it explicitly. If you do proper hashing on your secrets you may have to iterate over all tenants trying to find the matching one, which can become expensive really quickly. Or if you have a client that can act in different Tenants but using the same secret (though that example may be a bit unlikely, but anyway).

Anyway, I'd recommend using one of the methods mentioned in the RFC.

[hannes-danzl-eb]

I've not looked at this for a while now in regard to the fork. There were some issues getting this to prod but as far as I know they were mostly unrelated to this project. in any way, a variation is in prod for a few months now without any reports. I'll put some time aside over the next couple weeks to look into this a bit closer. Keep the comments coming.

[vpatov]

Coming back to this, I've done a lot of development on SCIM since. We are still using the contextvar pattern, and are experiencing no issues with it.

We went with the first option, having the tenant_id between the prefix and /v2/...

We also scope all of our SCIM URLs with the tenant_id, e.g. https://api.ourcompany.com/scim/<tenant_id>/Users.

Doing it implicitly via the authentication is technically an option, but I can imagine scenarios in which you'd want to do it explicitly.... Or if you have a client that can act in different Tenants but using the same secret (though that example may be a bit unlikely, but anyway).

I agree that having the tenant_id be explicitly present in the SCIM URL is better, because it also makes the caller's selection of tenant_id more explicit. Apologies for any confusion, I didn't mean to imply with my earlier post that one should use the context var + tenant info pattern instead of also prefixing the URLs.

If you do proper hashing on your secrets you may have to iterate over all tenants trying to find the matching one, which can become expensive really quickly.

That would be poor database design if this was the case. You always need to have an index on columns you use to perform lookups:

CREATE TABLE service_account_token (
    uid                      UUID PRIMARY KEY,
    tenant_uid               TEXT NOT NULL REFERENCES tenant(uid) ON DELETE CASCADE,
    secret_hash              TEXT NOT NULL CHECK (secret_hash ~ '^[0-9abcdef]{64}$'),
    service_account_uid      TEXT NOT NULL REFERENCES service_account(uid) ON DELETE CASCADE,
    FOREIGN KEY (tenant_uid, service_account_uid) REFERENCES service_account(tenant_uid, uid) ON DELETE CASCADE ON UPDATE CASCADE
);

-- Index to optimize lookups by secret_hash
CREATE INDEX idx_service_account_token_secret_hash ON service_account_token(secret_hash);

---

Some more context on how we do tenant isolation and auth:

We have a public-facing API-Gateway that performs authentication (looks up bearer tokens in an auth cache), and injects auth data into the request headers before forwarding the request to our SCIM server, so the auth data we need is available in those request headers, e.g (principal uid, tenant id, permissions, etc.)

Side note (unrelated to issue we're discussing but I'll nention it for posterity of good security posture): Backends can trust the request headers because only the API-Gateway is allowed to talk to the backends, and the request headers are signed for integrity

However, you don't need to have this architecture (API-Gateway that injects auth data into request headers) for that auth pattern to work. Your SCIM server can process up the token itself, and look up the tenant based on the token. Regardless of how you get the tenant info, you should 403 the request if the tenant_id from the tenant info does not match the tenant_id inside of the SCIM URL.

Here is what our check_auth function looks like now (slightly redacted):

class SCIMMiddleWare(BaseHTTPMiddleware):
    def __init__(self, app: FastAPI, backend: CloudSCIMBackend):
        super().__init__(app)
        self.backend = backend

    async def check_auth(self, request: Request):
 
       # in production, this looks at the request headers to get the tenant info (we call it auth_context)
       # in local dev, our backend gets the tenant info by looking it up via the token in the DB
        auth_context: ServiceAccountAuthenticatedContext | None = await auth_strategy(
            request
        )

        if auth_context is None:
            raise HTTPException(status_code=httpx.codes.UNAUTHORIZED)

        path_tenant_uid = re.match(
            rf"/scim/({settings.TENANT_UID_REGEX})/", request.url.path
        ).group(1)
        if path_tenant_uid != auth_context.principal.tenant_uid:
            raise HTTPException(
                status_code=httpx.codes.FORBIDDEN, detail="Unauthorized tenant_uid"
            )

        if Permissions.SCIM_API not in auth_context.principal.permissions:
            raise HTTPException(
                status_code=httpx.codes.FORBIDDEN,
                detail="Token is missing SCIM_API permission.",
            )
        return auth_context

Furthermore, all of our SCIM backend code assumes that a tenant_id is available, such that it can scope the DB queries to only that tenant. So for instance, if a user with uid abc exists in tenant 123, and I belong to tenant 234 and try to look up that user using my tenant SCIM URL, I'll simply get a 404 because that user doesn't exist in my tenant. For example:

curl 'https://api.ourcompany.com/scim/234/Users/abc' -H 'Authorization: Bearer <token for tenant 234>'
# NOT FOUND 404

curl 'https://api.ourcompany.com/scim/123/Users/abc' -H 'Authorization: Bearer <token for tenant 234>'
# FORBIDDEN 403

[hannes-danzl-eb]

Corporate policy doesn't allow me to share the full code, but the basic concept:

We subclass our own MultiTenantScimProvider. We add a TenantProvider class (subclass as needed) and then wire it into the Backend. This way you can create a multi-tenant setup without any changes to the original/base code. had to strip things back, so I hope I'm not missing anything right now. As mentioned, this is in prod since a while and seems to be working well.

class MultiTenantScimProvider(SCIMProvider):
    def __init__(self, backend):
        super().__init__(backend)
        self.log.setLevel(LOG_LEVEL)

    def wsgi_app(self, request: Request, environ):
        """ 
        override this function to enable request/response logging and 
        tenant information processing 
        """
        #pylint: disable=logging-fstring-interpolation
        if LOG_REQUESTS:
            self.log.info(f"Incoming request: {request.method} {request.url}")
            self.log.debug(f"Request headers: {dict(request.headers)}")
            if request.data:
                self.log.debug(f"Request payload: {request.get_data(as_text=True)}")

        is_multi_tenant = False
        if hasattr(self.backend, "is_multi_tenant") and self.backend.is_multi_tenant():
            is_multi_tenant = True

        response = None
        if not is_multi_tenant:
            response = super().wsgi_app(request, environ)
        else:
            # set the tenant info in the thread context
            if hasattr(self.backend, "set_tenant_info"):
                with self.backend.set_tenant_info(request):
                    response = super().wsgi_app(request, environ)

        # Log outgoing response details
        if response:
            if LOG_RESPONSES and response:
                self.log.debug(f"Outgoing response status: {response.status}")
                self.log.debug(f"Response headers: {dict(response.headers)}")
                if response.data:
                    self.log.debug(f"Response payload: {response.get_data(as_text=True)}")
            return response
        #pylint: enable=logging-fstring-interpolation
        
        raise Unauthorized

The most basic TenantProvider

class TenantInfo(BaseModel):
    id: str

_request_tenant_info: ContextVar[TenantInfo] = ContextVar(
    "request_tenant_info"
)

@contextmanager
def set_ctx_tenant_info(tenant_info):
    token = _request_tenant_info.set(tenant_info)
    try:
        yield
    finally:
        _request_tenant_info.reset(token)


def get_ctx_tenant_info():
    info = _request_tenant_info.get(None)
    if info is None:
        raise Unauthorized("Tenant info not set in context")
    return info

class TenantProvider:
    """ This class provides methods to handle auth and tokens. """
    def __init__(self):
        # Initialize logger in the base class
        self.logger = logging.getLogger("AUTHSERVICE")
        self.logger.setLevel(os.getenv("LOG_LEVEL", "INFO"))

    def set_tenant_info(self, tenant_info: TenantInfo):
        return set_ctx_tenant_info(tenant_info)
    
    def get_tenant_info(self) -> TenantInfo:
        return get_ctx_tenant_info()

    def get_jwt_payload(self, token: str) -> dict:
        """ need to be implemented in the child class """
        raise NotImplementedError("get_jwt_payload method not implemented")

Then we subclass our own Backend with a tenantprovider and the two required methods

class RestApiBackend(Backend):
    """A REST API backend for the SCIM provider."""

    def __init__(self, tenant_provider: TenantProvider):
        super().__init__()
        self.logger = logging.getLogger("BACKEND")
        self.logger.setLevel(os.getenv("LOG_LEVEL", "INFO"))
        self.tenant_provider = tenant_provider
        if tenant_provider:
            self.logger.info(f"Multi tenant mode using {type(self.tenant_provider).__name__} for authentication")
        else:
            self.logger.info("Single tenant mode")

    def is_multi_tenant(self) -> bool:
        return self.tenant_provider is not None

    def set_tenant_info(self, request: Request):
        """ Get the tenant id from the request. """
        if request and "Authorization" in request.headers:
            values = request.headers["Authorization"].split(" ")
            if len(values) > 1 and values[0].lower() == "bearer":
                try:
                    payload = self.tenant_provider.get_jwt_payload(values[1])
                    if payload:
                        info = TenantInfo(id=str(payload["tenant_id"]))
                        return self.tenant_provider.set_tenant_info(info)
                except: #pylint: disable=bare-except
                    self.logger.error("Failed to decode JWT token")

        return Unauthorized("Non-existing or invalid token")

... rest of your logic

Whenever you need the tenant_info in the backend, just use

tenant_info = self.tenant_provider.get_tenant_info()