How to use `AsyncSCIMClient` with `check_server`?

[vpatov]

Hello, thanks for writing a great series of libraries! I am having trouble understanding how to use the check_server functionality, while passing in the AsyncSCIMClient. My hunch is that this is not possible, because the check_server function flow seems to only be implemented synchronously. Is there anyway to get the benefit of the check_server test for my scim implementation, using an async client (and using the in-memory app object rather than hitting a server over network)?

Some more background: we have an ASGI FastAPI app. I originally tried mounting the SCIMProvider from scim2_server as WSGIMiddleware onto my FastAPI app, but since our app + DB layer is all async, I had to use async_to_sync in a bunch of places, and I ended up getting errors/issues with the event loop due to that. So instead, I opted to rewrite the SCIMProvider using an async interface (I replaced werkzeug with fastapi and starlette), while keeping all of the SCIM protocol logic + interfaces the same, such that I can use the same Backend interface, and leverage all the awesome work you folks have done with implementing SCIM. However, now I am having trouble finding good ways to leverage your library to also test the implementation, hence this issue.

Any help/guidance is much appreciated, thanks!

[azmeuk]

Hi.
Indeed scim2-tester is only synchronous at the moment, mostly because the async engine was not available in scim2-client when I started to implement scim2-tester. It would totally make sense for check_server to offer and async API too.

I am not sure how (and when) to tackle this, and make the tool both sync and async without too much code/logic duplication. Anyway, any help is welcome!

So instead, I opted to rewrite the SCIMProvider using an async interface (I replaced werkzeug with fastapi and starlette), while keeping all of the SCIM protocol logic + interfaces the same

Great! Is this something you may want to contribute upstream? Would be awesome to implement ASGI in scim2-server

[vpatov]

Great! Is this something you may want to contribute upstream? Would be awesome to implement ASGI in scim2-server

I'd love to, except currently my implementation suffers from exactly the problem you describe: too much code/logic duplication. It wouldn't be ideal for your library to be split along fastapi + ASGI / werkzeug + WSGI, with all of the logic agnostic to that also duplicated :/

To clarify, I manually rewrote this, and verified that the important logic is unaffected. However, it's essentially a duplicate of SCIMProvider, which uses an async backend CloudSCIMBackend. For example, here is a snippet:

# provider.py
import httpx
from fastapi import HTTPException, Request, Response
from fastapi.responses import JSONResponse
from pydantic import ValidationError
from scim2_models import ...
from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint
from werkzeug.http import unquote_etag # pure function doesn't need to be rewritten, doesn't necessarily need to be replaced, but ideally would be replaced with an equivalent fastapi import 

def continue_etag(request: Request, resource) -> bool:
    """Given a request and a resource, checks whether the ETag matches and allows continuing with the request.

    If the HTTP header "If-Match" is set, the request may only
    continue if the ETag matches. If the HTTP header "If-None-Match"
    is set, the request may only continue if the ETag does not
    match.

    Ported from scim2_server/provider.py's werkzeug implementation to fastapi.
    """
    cont = True
    resource_version, _ = unquote_etag(resource.meta.version)

    if_none_match = request.headers.get("if-none-match")
    if if_none_match:
        tokens = [token.strip() for token in if_none_match.split(",")]
        if resource_version in tokens:
            cont = False

    if_match = request.headers.get("if-match")
    if if_match:
        tokens = [token.strip() for token in if_match.split(",")]
        if resource_version not in tokens:
            cont = False

    return cont

...
class CloudSCIMProvider:
    """Provides SCIM protocol functionality.

    All functionality is ported from scim2_server.provider.SCIMProvider's werkzeug implementation to fastapi,
    with minimal modifications.
    """

    def __init__(self, backend: CloudSCIMBackend):
        self.backend = backend
        self.page_size = 50
        self.log = logging.getLogger(self.__class__.__name__)

    ....

    async def call_single_resource(
        self, request: Request, resource_endpoint: str, resource_id: str, **kwargs
    ) -> Response:
        find_endpoint = "/" + resource_endpoint
        resource_type = self.backend.get_resource_type_by_endpoint(find_endpoint)
        if not resource_type:
            raise HTTPException(status_code=httpx.codes.NOT_FOUND)

        match request.method:
            case "GET":
                if resource := await self.backend.get_resource(
                    resource_type.id, resource_id
                ):
                    if continue_etag(request, resource):
                        response_args = self.get_attrs_from_request(request)
                        self.adjust_location(request, resource)
                        return make_response(
                            resource.model_dump(
                                scim_ctx=Context.RESOURCE_QUERY_RESPONSE,
                                **response_args,
                            )
                        )
                    else:
                        return make_response(None, status=httpx.codes.NOT_MODIFIED)
                raise HTTPException(status_code=httpx.codes.NOT_FOUND)

# api.py

backend = CloudSCIMBackend()
scim_provider = CloudSCIMProvider(backend)
for schema in load_default_schemas().values():
    scim_provider.register_schema(schema)
for resource_type in load_default_resource_types().values():
    scim_provider.register_resource_type(resource_type)


app = FastAPI(middleware=[Middleware(SCIMMiddleWare, backend=backend)])

...

@app.get("/{resource_endpoint}")
async def resource_get(request: Request, resource_endpoint: str):
    return await scim_provider.call_resource(
        request, resource_endpoint=resource_endpoint
    )

# middleware.py

from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint


# `set_auth_context` is an async context manager that uses ContextVar to make auth_context
# available to every part of the code that needs it e.g. provider, backend, without needing to
# explicitly pass it down in the constructor/function args

class SCIMMiddleWare(BaseHTTPMiddleware):
    def __init__(self, app, backend: CloudSCIMBackend):
        super().__init__(app)
        self.backend = backend
        self.log = logging.getLogger(self.__class__.__name__)

    async def check_auth(self, request: Request):
        # this function implementation is highly specific to our backend
        pass

    async def dispatch(self, request: Request, call_next: RequestResponseEndpoint):
        try:
            # TODO redirect URLs that end with .scim to the non-suffixed URL

            auth_context = None
            if "ServiceProviderConfig" not in request.url.path:
                auth_context = await self.check_auth(request)

            async with self.backend, set_auth_context(auth_context):
                response = await call_next(request)

            if "location" not in response.headers:
                response.headers["location"] = str(request.url)

            return response

        except HTTPException as e:
            self.log.exception(e)
            return make_error(Error(status=e.status_code, detail=e.detail))
        except SCIMException as e:
            self.log.exception(e)
            return make_error(e.scim_error)
        except ValidationError as e:
            self.log.exception(e)
            return make_error(Error(status=httpx.codes.BAD_REQUEST, detail=str(e)))
        except Exception as e:
            self.log.exception(e)
            sentry_sdk.capture_exception(e)
            return make_error(Error(status=httpx.codes.INTERNAL_SERVER_ERROR))

I am not aware of solid and well-tested strategies for having one implementation provide both an async and sync interface. In my experience,async_to_sync historically has had weird gotchas/problems.

However, if you are interested in the complete fastapi ASGI implementation that I reference above, despite the code duplication, I can put together a PR (that is stripped of all of our company-specific logic) sometime in the near future.

---

UPDATE: I am also finding some problems with my implementation as-is, for example exception handling works differently in fastapi, and the exception handlers I have in the dispatch method are not actually being triggered if the endpoint processing raises an exception (from the call to call_next), starlette takes care of them elsewhere. So I will need to change how I'm doing this. I discovered this bug, by manually using the AsyncSCIMClient.query method in some hand-written tests (since I can't use check_server at the moment), and seeing that it was failing during the check_response portion.