Overview#

This tutorial builds a SCIM server over the tables of an existing application. The application is an intranet that keeps accounts, their email addresses, and teams. At the end, the server serves the accounts as SCIM users and the teams as SCIM groups, and passes the checks of scim2-tester.

The tutorial is for Python developers whose application stores its data with the SQLAlchemy ORM, version 2.0. It assumes familiarity with SQLAlchemy models and sessions, and with the users and groups of SCIM (RFC 7643 §4). It runs the server without a web framework. To serve it from one, read Integrate a web framework. To commit its changes, read Manage the transactions.

Install scim2-sqlalchemy. It brings SQLAlchemy, scim2-server and scim2-models along:

$ pip install scim2-sqlalchemy

Start from the tables#

The intranet already has its SQLAlchemy models. scim2-sqlalchemy uses them as they are. An account has a login, a name, a password hash, email addresses and teams:

class Account(Base):
    __tablename__ = "accounts"

    id: Mapped[uuid.UUID] = mapped_column(primary_key=True, default=uuid.uuid4)
    external_id: Mapped[str | None]
    login: Mapped[str] = mapped_column(unique=True)
    first_name: Mapped[str | None]
    last_name: Mapped[str | None]
    active: Mapped[bool | None] = mapped_column(default=True)
    office: Mapped[str | None]
    password_hash: Mapped[str | None]
    created_at: Mapped[datetime.datetime]
    updated_at: Mapped[datetime.datetime]
    revision: Mapped[int] = mapped_column()

    emails: Mapped[list["EmailAddress"]] = relationship(
        cascade="all, delete-orphan", order_by="EmailAddress.id"
    )
    teams: Mapped[list["Team"]] = relationship(
        secondary=team_members, back_populates="members"
    )

    __mapper_args__ = {"version_id_col": revision}

    @hybrid_property
    def full_name(self):
        return f"{self.first_name or ''} {self.last_name or ''}".strip() or None

    @full_name.expression
    def full_name(cls):
        return func.nullif(
            func.trim(
                func.coalesce(cls.first_name, "")
                + " "
                + func.coalesce(cls.last_name, "")
            ),
            "",
        )

    @hybrid_property
    def password(self):
        return self.password_hash

    @password.inplace.setter
    def _password_setter(self, value):
        self.password_hash = None if value is None else hash_password(value)

The models need four things for SCIM:

  • a primary key unique across the accounts and the teams, here a UUID. Per RFC 7643 §3.1, two resources never share an id, even of different types;

  • a column for the identifier the identity provider gives, here external_id;

  • a creation date and a modification date, here created_at and updated_at;

  • a version counter, declared as the version_id_col of the mapper, here revision. SQLAlchemy increments it on every write, and SCIM clients send it back in their If-Match headers.

The email addresses and the teams live in tables of their own:

class EmailAddress(Base):
    __tablename__ = "email_addresses"

    id: Mapped[int] = mapped_column(primary_key=True)
    account_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("accounts.id"))
    address: Mapped[str]
    kind: Mapped[str | None]
    preferred: Mapped[bool | None]

class Team(Base):
    __tablename__ = "teams"

    id: Mapped[uuid.UUID] = mapped_column(primary_key=True, default=uuid.uuid4)
    external_id: Mapped[str | None]
    name: Mapped[str]
    created_at: Mapped[datetime.datetime]
    updated_at: Mapped[datetime.datetime]
    revision: Mapped[int] = mapped_column()

    members: Mapped[list[Account]] = relationship(
        secondary=team_members, back_populates="teams"
    )

    __mapper_args__ = {"version_id_col": revision}

Map the accounts to users#

A ResourceMapping stores a SCIM model in a SQLAlchemy model. It maps the path of each SCIM attribute to a column. The following mapping stores the User resources in the accounts table:

USERS = ResourceMapping(
    User,
    Account,
    {
        "id": Account.id,
        "externalId": Account.external_id,
        "userName": Account.login,
        "name.givenName": Account.first_name,
        "name.familyName": Account.last_name,
        "displayName": Account.full_name,
        "active": Account.active,
        "password": Attribute(Account.password, readable=False),
        "emails": Many(
            Account.emails,
            {
                "value": EmailAddress.address,
                "type": EmailAddress.kind,
                "primary": EmailAddress.preferred,
            },
        ),
        "groups": Link(Account.teams, "Group"),
        "meta.created": Account.created_at,
        "meta.lastModified": Account.updated_at,
    },
    version=Account.revision,
)

Each value of the mapping says how the attribute is stored:

  • externalId, userName, name.givenName and active are plain columns. externalId holds the identifier the identity provider gives to the user.

  • displayName is a hybrid_property without setter: the storage reads it, filters and sorts on it, and never writes it.

  • password is a property whose setter hashes the password. The Attribute around it makes it write-only: the storage writes it, and never reads it back.

  • emails is a Many: the email addresses belong to the account, and writing the user replaces them.

  • groups is a Link: the teams are other resources, of the Group resource type.

  • meta.created and meta.lastModified are the dates the storage fills.

The mapping checks every path against the User model when the module is imported. A misspelled path raises a ValueError at once.

Map the teams to groups#

The teams become Group resources. Their members link back to the users:

GROUPS = ResourceMapping(
    Group,
    Team,
    {
        "id": Team.id,
        "externalId": Team.external_id,
        "displayName": Team.name,
        "members": Link(Team.members, "User"),
        "meta.created": Team.created_at,
        "meta.lastModified": Team.updated_at,
    },
    version=Team.revision,
)

Describe the service#

A ScimProvider describes the service: the schemas of its resources, its resource types, and the features it supports. Build it from the mappings: the schemas of each mapping, and a resource type built from the model of each mapping, which lists the extensions the mapping stores. schemas() restricts each schema to the attributes the mapping stores, so that clients only send what the server keeps:

def create_provider(policy=None):
    return ScimProvider.from_discovery(
        [*USERS.schemas(), *GROUPS.schemas()],
        [
            ResourceType.from_resource(USERS.model),
            ResourceType.from_resource(GROUPS.model),
        ],
        config=load_default_service_provider_config(),
        policy=policy,
    )

Create the storage#

A SqlAlchemyStorage reads and writes the resources with a SQLAlchemy Session. It receives the mapping of each resource type, a function returning the current session, and the provider:

def create_storage(session, provider):
    return SqlAlchemyStorage(
        {"User": USERS, "Group": GROUPS}, lambda: session, provider=provider
    )

Create the tables in an in-memory SQLite database, and open a session:

>>> from sqlalchemy import create_engine
>>> from sqlalchemy.orm import Session
>>> from intranet.models import Base
>>> engine = create_engine("sqlite://")
>>> Base.metadata.create_all(engine)
>>> session = Session(engine)

Serve the storage with the WSGI application of scim2-server, which handles the SCIM protocol:

>>> from scim2_server.applications.wsgi import WSGIApplication
>>> from intranet.scim import create_provider
>>> from intranet.scim import create_storage
>>> provider = create_provider()
>>> app = WSGIApplication(create_storage(session, provider), provider)

Serve requests#

The application now answers SCIM requests. The examples send them with the test client of Werkzeug, which calls the application without running a server. Create a user:

>>> from werkzeug.test import Client
>>> client = Client(app)
>>> USER = "urn:ietf:params:scim:schemas:core:2.0:User"
>>> response = client.post(
...     "/v2/Users",
...     json={
...         "schemas": [USER],
...         "userName": "bjensen",
...         "name": {"givenName": "Barbara", "familyName": "Jensen"},
...         "emails": [{"value": "bjensen@example.com", "type": "work", "primary": True}],
...         "password": "t1meMa$heen",
...     },
... )
>>> response.status_code
201
>>> user = response.json
>>> user["displayName"], user["active"], "password" in user
('Barbara Jensen', True, False)

The response comes from the new row of the accounts table. displayName comes from the hybrid property, and active from the default of the active column. The password is stored hashed:

>>> import uuid
>>> from intranet.models import Account
>>> session.get(Account, uuid.UUID(user["id"])).password_hash.count(":")
1

Create a group with this user as a member. The member links to the user, with its URL:

>>> GROUP = "urn:ietf:params:scim:schemas:core:2.0:Group"
>>> response = client.post(
...     "/v2/Groups",
...     json={"schemas": [GROUP], "displayName": "Admins", "members": [{"value": user["id"]}]},
... )
>>> response.json["members"]
[{'value': '...', '$ref': 'http://localhost/v2/Users/...', 'display': 'Barbara Jensen'}]

Search the users. The storage turns the filter and the sort into a SQL query, so the database only returns the matching rows:

>>> response = client.get(
...     "/v2/Users",
...     query_string={"filter": 'emails[type eq "work"] and groups pr', "sortBy": "userName"},
... )
>>> response.json["totalResults"], response.json["Resources"][0]["userName"]
(1, 'bjensen')

An attribute the server does not store is refused, rather than lost:

>>> response = client.post(
...     "/v2/Users",
...     json={"schemas": [USER], "userName": "jsmith", "phoneNumbers": [{"value": "555-1234"}]},
... )
>>> response.status_code, response.json["detail"]
(400, 'Extra inputs are not permitted: phoneNumbers')

Check the server#

scim2-tester checks that a SCIM server complies with the RFCs. It sends the requests an identity provider would send, and checks each response. The WSGI engine of scim2-client calls the application directly, without a network:

$ pip install scim2-tester

Run the checks:

>>> from scim2_client.engines.wsgi import WSGISCIMClient
>>> from scim2_tester import Status
>>> from scim2_tester import check_server
>>> scim_client = WSGISCIMClient(app, base_url="http://localhost/v2", provider=provider)
>>> results = check_server(scim_client)
>>> [result.title for result in results if result.status not in (Status.SUCCESS, Status.SKIPPED)]
[]

Close the session and the engine when the application stops:

>>> session.close()
>>> engine.dispose()

Next steps#

The server now serves the accounts and the teams of the intranet: