diff --git a/docs/conf_sso_test_msscolab.rst b/docs/conf_sso_test_msscolab.rst index 1927640e5..0a3d51810 100644 --- a/docs/conf_sso_test_msscolab.rst +++ b/docs/conf_sso_test_msscolab.rst @@ -8,7 +8,10 @@ Testing IDP (`mslib/msidp`) is specifically designed for testing the Single Sign Here is documentation that explains the configuration of the MSS Colab Server with the testing IdP. .. warning:: - When running publicly rather than in development, you should not use the built-in development server ( msidp / idp.py ). + msidp is for development and tests only. Never run it where others can reach it, and never let an MSColab + server that others use trust it: its users and passwords are the ones in ``mslib/msidp/idp_user.py``, which + everyone knows, and it lacks protections of a real identity provider. For example it accepts unsigned requests, + has no CSRF protection on its login form and redirects to any ``redirect_uri``. It prints this warning on start. The development server is provided by MSS for convenience, but is not designed to be particularly efficient, stable, or secure. @@ -127,5 +130,7 @@ When migrations finished, you can start mscolab server using the following comm $ msui -* Login with identity provider through Qt Client application. +* Login with identity provider through Qt Client application. After the login at msidp the browser shows a + one-time code, enter it in msui. If an MSColab account with the email of the msidp user exists already, msui asks + for its password once to link the msidp login to it. * To log in to the mscolab server through the identity provider, you can use the credentials specified in the ``USERS`` and ``PASSWD`` section of the ``MSS/mslib/msidp/idp_user.py`` file. Look for the relevant section in the file to find the necessary login credentials. diff --git a/docs/samples/config/mscolab/mss_saml2_backend.yaml.sample b/docs/samples/config/mscolab/mss_saml2_backend.yaml.sample index 09711f2ac..7b7c1e159 100644 --- a/docs/samples/config/mscolab/mss_saml2_backend.yaml.sample +++ b/docs/samples/config/mscolab/mss_saml2_backend.yaml.sample @@ -52,7 +52,8 @@ config: want_response_signed: true want_assertion_signed: true allow_unknown_attributes: true - allow_unsolicited: true + # only responses to a login request of this server + allow_unsolicited: false endpoints: assertion_consumer_service: - [http://localhost:8083/localhost_test_idp/acs/post, 'urn:oasis:names:tc:SAML:2.0:bindings:HTTP-POST'] @@ -106,7 +107,7 @@ config: # want_response_signed: true # want_assertion_signed: true # allow_unknown_attributes: true - # allow_unsolicited: true + # allow_unsolicited: false # endpoints: # assertion_consumer_service: # - [http://localhost:8083/idp2/acs/post, 'urn:oasis:names:tc:SAML:2.0:bindings:HTTP-POST'] diff --git a/docs/samples/config/mscolab/setup_saml2_backend.py.sample b/docs/samples/config/mscolab/setup_saml2_backend.py.sample index edc4a8963..3c47f81d7 100644 --- a/docs/samples/config/mscolab/setup_saml2_backend.py.sample +++ b/docs/samples/config/mscolab/setup_saml2_backend.py.sample @@ -17,6 +17,9 @@ class setup_saml2_backend: 'idp_identity_name': 'localhost_test_idp', # make sure to use underscore for the blanks 'idp_data': { 'idp_name': 'Testing Identity Provider', # this name is used on the Login page to connect to the Provider. + # the attribute that identifies a user: it must not change and must never be given to someone else. + # Without it the IdP must send a persistent NameID. + 'user_id_attribute': 'eduPersonPrincipalName', } }, diff --git a/docs/sso_via_saml_mscolab.rst b/docs/sso_via_saml_mscolab.rst index c531ff305..3811038f5 100644 --- a/docs/sso_via_saml_mscolab.rst +++ b/docs/sso_via_saml_mscolab.rst @@ -102,6 +102,7 @@ In this implementation, as we are enabling only one IdP, there is no need to con 'idp_identity_name': 'localhost_test_idp', # make sure to use underscore for the blanks 'idp_data': { 'idp_name': 'Testing Identity Provider', # this name is used on the Login page to connect to the Provider + 'user_id_attribute': 'eduPersonPrincipalName', # the attribute that identifies a user } }, ] @@ -116,6 +117,12 @@ In this implementation, as we are enabling only one IdP, there is no need to con Idp_name refers to the name of the Identity Provider that will be displayed in the MSColab server web interface for end users to select when configuring SSO. + User_id_attribute is the SAML attribute that identifies a user of this IdP, e.g. ``eduPersonPrincipalName`` + or ``urn:oasis:names:tc:SAML:attribute:subject-id``. Its value must not change and must never be given to + someone else. Without ``user_id_attribute`` the IdP must send a persistent NameID + (``urn:oasis:names:tc:SAML:2.0:nameid-format:persistent``). MSColab refuses a login without such an + identifier. The email address is no identifier: it can change, and an IdP can assert any address. + Establish pysaml2, Saml2Client for the MSColab server ##################################################### @@ -189,7 +196,7 @@ Please refer the yaml file template (`mss_saml2_backend.yaml.sample`) in the dir want_response_signed: true want_assertion_signed: true allow_unknown_attributes: true - allow_unsolicited: true + allow_unsolicited: false endpoints: assertion_consumer_service: - [http://localhost:8083/idp2/acs/post, 'urn:oasis:names:tc:SAML:2.0:bindings:HTTP-POST'] @@ -221,7 +228,48 @@ While the core purpose of IdPs is to authenticate users and provide information Guide to IDP Configuration -------------------------- -In the SSO process through the MSColab server, the username is obtained as ``givenName``, and the email address is obtained as ``email``. Therefore, when configuring the IdP, it is necessary to configure it accordingly to ensure the correct return of the givenName attribute and the email address along with the SAML response. +In the SSO process through the MSColab server, the username is obtained as ``givenName`` (or ``displayName``), and the email address is obtained as ``email`` (or ``mail``). Therefore, when configuring the IdP, it is necessary to configure it accordingly to ensure the correct return of the givenName attribute and the email address along with the SAML response. The IdP must also send the attribute of ``user_id_attribute`` or a persistent NameID, see above. + +How a SAML login finds its account +---------------------------------- + +MSColab links an account to SAML identities: the entity id of the IdP together with the value of +``user_id_attribute`` (or the persistent NameID). A login by a linked identity logs in to its account, also when the +IdP sends another email address than before. + +* The first login by an identity that isn't linked yet gets a new account with the email address and the given name + that the IdP sends. The given name becomes the username; if it is taken, contains ``@`` or is longer than 255 + characters, the part of the email address before the ``@`` does. The browser shows why a login is refused: no valid + email address, or both usernames can't be used, or several accounts have the email address in different upper and + lower case. +* If an account with that email address exists already, in any upper and lower case, the login is refused until its + owner links the identity: msui asks for the password of that account once and links the identity right after the + password login. +* Accounts of MSColab before 12.0.0 that logged in by SAML have no linked identity yet. Such an account is linked on + its next login by email, if it logged in by the same configured IdP before and the response comes from the IdP that + MSColab sent the login request to. A response of another IdP of the same metadata, e.g. of a federation, needs the + link by the owner, like any other existing account. +* With ``MAIL_ENABLED`` an account whose email address isn't confirmed can't log in by SAML either; msui tells so, and + the code works again once the address is confirmed. +* A SAML login never changes the password of an account. Accounts created by a SAML login have a random password; their + owners can set one by a password reset. A password reset unlinks the SAML identities of the account: whoever resets + the password owns the email address. The next SAML login links the identity again, with the new password. +* Without ``user_id_attribute`` an IdP must send a persistent NameID. The server warns at its start if neither is set + up for a configured IdP: every SAML login by it would be refused. + +The login in msui +----------------- + +After the login at the IdP the browser shows a one-time code. The user enters it in msui, which exchanges it for a +login token. The code works once and for 20 minutes; the IdP has to answer the login request within 10 minutes. The +server accepts only responses to login requests it sent, each once, so ``allow_unsolicited`` must be ``false``; a +login started at the IdP (IdP-initiated) doesn't work. The requests are stored in the database, so every process of +the server accepts the response. Anyone can start a login, so the server keeps at most 1000 unfinished ones; further +logins are refused with "503 Service Unavailable" until some of them expire. With ``ENABLE_BASIC_HTTP_AUTHENTICATION`` +the browser needs the basic authentication of the server to start a login, and msui to exchange the code, as for a +password login. An assertion that isn't signed counts only if the signed response around it comes from the same IdP. +msui 12.0.0 also logs in by SAML to servers before 12.0.0; msui before 12.0.0 can't log in by SAML to a 12.0.0 +server. 3. Configuration example through Keycloak 13.0.1 @@ -469,6 +517,10 @@ Configuration in MSColab settings for Keycloak .. note:: Make sure to insert idp_identity_name as above ('key_cloak_v_13'), which used in this example. + This example has no ``user_id_attribute``: set the Name ID Format of the client in Keycloak to + ``persistent`` and ``name_id_format`` in `mss_saml2_backend.yaml` to + ``urn:oasis:names:tc:SAML:2.0:nameid-format:persistent``, then the NameID identifies the user. + Configuration mss_saml2_backend.yaml file Create your mss_saml2_backend.yaml file in your ``SSO_DIR``. @@ -528,7 +580,7 @@ Configuration mss_saml2_backend.yaml file want_response_signed: true want_assertion_signed: true allow_unknown_attributes: true - allow_unsolicited: true + allow_unsolicited: false endpoints: assertion_consumer_service: - [http://localhost:8083/keycloak_idp/acs/post, 'urn:oasis:names:tc:SAML:2.0:bindings:HTTP-POST'] diff --git a/mslib/mscolab/CLAUDE.md b/mslib/mscolab/CLAUDE.md index b0002f8e3..21c79438f 100644 --- a/mslib/mscolab/CLAUDE.md +++ b/mslib/mscolab/CLAUDE.md @@ -70,6 +70,13 @@ and `servers-without-qt` contracts in setup.cfg. `logout_everywhere`) sets a new one and invalidates all earlier tokens. Email-link tokens use `generate_confirmation_token(email, purpose)` with their own salt per purpose, and links in emails are built with `public_url_for` from `PUBLIC_URL`, never `_external=True`. +- A SAML login is accepted by the linked `IdpIdentity` (IdP entity id + configured `user_id_attribute` or + persistent NameID), never by email alone, and never changes a password. Only responses to a request in + `saml_logins` are accepted, once, at most `MAX_SAML_LOGINS` open at a time; msui exchanges the one-time code at + `idp_login_auth` (see `auth.idp_login`). The issuer is the one of the signed assertion (of an unsigned one only if + the signed response has the same issuer). A pre-12 account is linked by email only for the IdP the request went + to. The code is used up in the same transaction that creates or links the account, so a failure leaves it valid. + The ACS routes are added to the app in `init_saml`: a blueprint can't change after its first registration. - No module level state that belongs to one app (app, db engine, SocketIO, HTTPBasicAuth, managers). It lives on the app: `app.config` or `app.extensions['sockio' | 'cm' | 'fm' | 'basic_auth']`, read via `current_app`. diff --git a/mslib/mscolab/api/endpoints.py b/mslib/mscolab/api/endpoints.py index 9cfb240a9..8aa7bef70 100644 --- a/mslib/mscolab/api/endpoints.py +++ b/mslib/mscolab/api/endpoints.py @@ -57,5 +57,6 @@ STATUS = "status" REGISTER = "register" IDP_LOGIN_AUTH = "idp_login_auth" +IDP_LINK = "idp_link" MESSAGES = "messages" MESSAGE_ATTACHMENT = "message_attachment" diff --git a/mslib/mscolab/api/schemas.py b/mslib/mscolab/api/schemas.py index 871ea0ffd..910c2e77b 100644 --- a/mslib/mscolab/api/schemas.py +++ b/mslib/mscolab/api/schemas.py @@ -469,9 +469,7 @@ def from_text(cls, text): @dataclass class LoginRequest: - """POST endpoints.TOKEN -- log in (email+password), or complete an IDP - login by posting the IDP-issued token as the "password". - """ + """POST endpoints.TOKEN -- log in (email+password).""" email: str password: str @@ -1096,14 +1094,15 @@ def from_text(cls, text): @dataclass class IdpLoginAuthRequest: - """POST endpoints.IDP_LOGIN_AUTH -- submit the SAML-issued token to - complete an IDP login. + """POST endpoints.IDP_LOGIN_AUTH -- exchange the one-time code that the + browser shows after a SAML login for a login token. Unlike every other route migrated so far, the client sends this as a real JSON body (`requests.post(url, json=...)`), not form-encoded data, and the server reads it via `request.get_json()`, not `request.form` -- so this is `to_json_data()`/`from_json_data()`, not - `to_form_data()`/`from_form()`. + `to_form_data()`/`from_form()`. The field is called token for + compatibility, it holds the code. """ token: str @@ -1118,21 +1117,22 @@ def from_json_data(cls, data): @dataclass class IdpUserInfo: - """User info embedded in IdpLoginAuthResponse -- a different shape than - UserInfo (username/id/emailid here, vs. username/id/fullname there; no - route shares this exact shape with any other migrated so far). + """User info embedded in IdpLoginAuthResponse -- UserInfo plus emailid, + so that the response is also a LoginResponse for msui's after_login. """ username: str id: int # noqa: A003 emailid: str + fullname: str = "" def to_dict(self): - return {"username": self.username, "id": self.id, "emailid": self.emailid} + return {"username": self.username, "id": self.id, "emailid": self.emailid, "fullname": self.fullname} @classmethod def from_dict(cls, data): - return cls(username=data["username"], id=data["id"], emailid=data["emailid"]) + return cls(username=data["username"], id=data["id"], emailid=data["emailid"], + fullname=data.get("fullname") or "") @dataclass @@ -1141,30 +1141,86 @@ class IdpLoginAuthResponse: The success path returns via a bare json.dumps (text/html Content-Type, same pattern as create_operation/get_operation_by_id) with - {"success": True, "token": ..., "user": {...}}. Every failure path - (invalid/expired token, no matching user, or a malformed request - raising TypeError) returns via flask.jsonify (application/json) with - just {"success": False} and HTTP 401 -- to_dict() lets the handler keep - using jsonify() for that path without silently changing its - Content-Type, mirroring DeleteOperationResponse's to_dict()/to_text() - split. + {"success": True, "token": ..., "user": {...}}; token is the login token, + the same as the one of POST endpoints.TOKEN. Every failure path returns + via flask.jsonify (application/json) with {"success": False}: HTTP 401 + for an invalid, expired or used code, HTTP 409 with link_required and + the emailid of the account if the SAML identity isn't linked to an + account yet and its email belongs to one -- the user logs in to that + account and links the identity by POST endpoints.IDP_LINK with the same + code. HTTP 403 with a message for the user if the login is refused, + e.g. for an account whose email isn't confirmed. to_dict() lets the + handler keep using jsonify() for these paths. + + Servers before 12.0.0 answer success with a one-time password in token; + a 12.0.0 server marks its login token with "login_token": true. + legacy_password, a success without that mark, tells msui to log in with + the password at endpoints.TOKEN, as the older servers expect. """ success: bool token: Optional[str] = None user: Optional[IdpUserInfo] = None + link_required: bool = False + emailid: Optional[str] = None + message: Optional[str] = None + legacy_password: bool = False def to_text(self): - return json.dumps({"success": self.success, "token": self.token, "user": self.user.to_dict()}) + return json.dumps({"success": self.success, "token": self.token, "login_token": True, + "user": self.user.to_dict()}) def to_dict(self): - return {"success": self.success} + data = {"success": self.success} + if self.link_required: + data.update(link_required=True, emailid=self.emailid) + if self.message is not None: + data.update(message=self.message) + return data @classmethod def from_text(cls, text): data = json.loads(text) - user = IdpUserInfo.from_dict(data["user"]) if "user" in data else None - return cls(success=data["success"], token=data.get("token"), user=user) + user = IdpUserInfo.from_dict(data["user"]) if data.get("user") else None + return cls(success=data["success"], token=data.get("token"), user=user, + link_required=bool(data.get("link_required", False)), emailid=data.get("emailid"), + message=data.get("message"), + legacy_password=bool(data.get("success")) and data.get("login_token") is not True) + + +@dataclass +class IdpLinkRequest: + """POST endpoints.IDP_LINK -- link the SAML identity of a one-time code, + which endpoints.IDP_LOGIN_AUTH refused with link_required, to the logged + in user (token is the login token, as for every other route). + """ + + token: str + code: str + + def to_form_data(self): + return {"token": self.token, "code": self.code} + + @classmethod + def from_form(cls, form): + return cls(token=form.get("token"), code=form.get("code")) + + +@dataclass +class IdpLinkResponse: + """Response body for POST endpoints.IDP_LINK (or the body "False" for an + invalid login token, as for every route of verify_user).""" + + success: bool + + def to_dict(self): + return {"success": self.success} + + @classmethod + def from_text(cls, text): + if text == AUTH_FAILED_TEXT: + return cls(success=False) + return cls(success=bool(json.loads(text).get("success"))) @dataclass diff --git a/mslib/mscolab/app/__init__.py b/mslib/mscolab/app/__init__.py index 77c5cb9d5..caa198612 100644 --- a/mslib/mscolab/app/__init__.py +++ b/mslib/mscolab/app/__init__.py @@ -271,9 +271,10 @@ def repair_schema(): """ from alembic.migration import MigrationContext from alembic.operations import Operations - from mslib.mscolab.migrations.schema_repair import (AUTOINCREMENT_TABLES, add_fullname, add_token_nonce, - fix_postgresql_sequences, has_autoincrement_ids, - has_column, has_dangling_references, + from mslib.mscolab.migrations.schema_repair import (AUTOINCREMENT_TABLES, add_fullname, add_saml_tables, + add_token_nonce, fix_postgresql_sequences, + has_autoincrement_ids, has_column, + has_dangling_references, has_table, sqlite_foreign_keys_off, use_autoincrement_ids) from mslib.mscolab.models import db @@ -294,6 +295,10 @@ def repair_schema(): logging.warning("The database misses users.token_nonce, it ran an earlier development state of the " "12.0.0 migration. Adding it now; every user has to log in again.") add_token_nonce(Operations(MigrationContext.configure(conn)), conn) + if not (has_table(conn, "idp_identities") and has_table(conn, "saml_logins")): + logging.warning("The database misses the tables of SAML logins, it ran an earlier development state of " + "the 12.0.0 migration. Adding them now.") + add_saml_tables(Operations(MigrationContext.configure(conn)), conn) if has_dangling_references(conn): # Only a database that ran an earlier development state of the 12.0.0 migration, which repairs them: # a server start doesn't delete rows. Until they are repaired, deleting their user or operation fails. diff --git a/mslib/mscolab/auth.py b/mslib/mscolab/auth.py index 2c7312ad0..8f0431b18 100644 --- a/mslib/mscolab/auth.py +++ b/mslib/mscolab/auth.py @@ -26,20 +26,24 @@ """ +import collections.abc import datetime import functools import hashlib import hmac import logging +import secrets import email_validator import sqlalchemy from flask import current_app, request, abort, g, url_for from itsdangerous import URLSafeTimedSerializer, BadSignature +from saml2.saml import NAMEID_FORMAT_PERSISTENT from mslib.mscolab.api.schemas import TOKEN_HEADER +from mslib.mscolab.app import db from mslib.mscolab.conf import setup_saml2_backend -from mslib.mscolab.models import User, fits_column +from mslib.mscolab.models import IdpIdentity, SamlLogin, User, fits_column def optional_auth(f): @@ -153,8 +157,7 @@ def wrapper(*args, **kwargs): # email confirmation, can't be used for another, e.g. a password reset. EMAIL_CONFIRMATION = "email-confirmation" PASSWORD_RESET = "password-reset" -IDP_LOGIN = "idp-login" -_TOKEN_PURPOSES = (EMAIL_CONFIRMATION, PASSWORD_RESET, IDP_LOGIN) +_TOKEN_PURPOSES = (EMAIL_CONFIRMATION, PASSWORD_RESET) def _serializer(purpose): @@ -178,7 +181,7 @@ def confirm_token(token, purpose, expiration=3600): def generate_confirmation_token(email, purpose): """ - A signed token with email for purpose, one of EMAIL_CONFIRMATION, PASSWORD_RESET and IDP_LOGIN + A signed token with email for purpose, one of EMAIL_CONFIRMATION and PASSWORD_RESET """ return _serializer(purpose).dumps(email) @@ -225,40 +228,357 @@ def public_url_for(endpoint, **values): return current_app.config['PUBLIC_URL'].rstrip("/") + url_for(endpoint, **values) +def find_idp_config(idp_identity_name): + """The configured IdP idp_identity_name of setup_saml2_backend.CONFIGURED_IDPS, None if there is none""" + for idp_config in setup_saml2_backend.CONFIGURED_IDPS: + if idp_config['idp_identity_name'] == idp_identity_name: + return idp_config + return None + + def get_idp_entity_id(selected_idp): """ Finds the entity_id from the configured IDPs :return: the entity_id of the idp or None """ - for config in setup_saml2_backend.CONFIGURED_IDPS: - if selected_idp == config['idp_identity_name']: - idps = config['idp_data']['saml2client'].metadata.identity_providers() - only_idp = idps[0] - entity_id = only_idp - return entity_id - return None + idp_config = find_idp_config(selected_idp) + if idp_config is None: + return None + return idp_config['idp_data']['saml2client'].metadata.identity_providers()[0] + + +# The IdP must answer a SAML login within SAML_REQUEST_LIFETIME seconds, msui must exchange its code within +# IDP_LOGIN_CODE_LIFETIME seconds after that +SAML_REQUEST_LIFETIME = 600 +IDP_LOGIN_CODE_LIFETIME = 1200 +# Anyone can start a SAML login, the server keeps at most this many unfinished ones +MAX_SAML_LOGINS = 1000 + + +def _now(): + return datetime.datetime.now(tz=datetime.timezone.utc) + + +def _saml_requests(idp_identity_name, request_id): + # the login request_id sent to the IdP idp_identity_name, if the IdP can still answer it + return SamlLogin.query.filter( + SamlLogin.request_id == str(request_id), + SamlLogin.idp_identity_name == str(idp_identity_name), + SamlLogin.code_hash.is_(None), + SamlLogin.created_on >= _now() - datetime.timedelta(seconds=SAML_REQUEST_LIFETIME)) -def create_or_update_idp_user(email, username, token, authentication_backend): +def remember_saml_request(request_id, idp_identity_name): """ - Creates or updates an idp user in the system based on the provided email, - username, token, and authentication backend. - :param email: idp users email - :param username: idp users username - :param token: authentication token - :param authentication_backend: authenticated identity providers name - :return: bool : query success or not + Stores the id of an AuthnRequest sent to the IdP idp_identity_name: only a response to it is accepted, and once + + Returns False and stores nothing if MAX_SAML_LOGINS logins are unfinished, until some of them expire. """ - fm = current_app.extensions['fm'] - user = User.query.filter_by(emailid=email).first() - if not user: - # using an IDP for a new account/profile, e-mail is already verified by the IDP - confirm_time = datetime.datetime.now(tz=datetime.timezone.utc) + datetime.timedelta(seconds=1) - user = User(email, username, password=token, confirmed=True, confirmed_on=confirm_time, - authentication_backend=authentication_backend) - result = fm.modify_user(user, action="create") + # expired requests and codes + SamlLogin.query.filter(sqlalchemy.or_( + sqlalchemy.and_(SamlLogin.code_hash.is_(None), + SamlLogin.created_on < _now() - datetime.timedelta(seconds=SAML_REQUEST_LIFETIME)), + SamlLogin.created_on < _now() - datetime.timedelta(seconds=IDP_LOGIN_CODE_LIFETIME), + )).delete(synchronize_session=False) + if SamlLogin.query.count() >= MAX_SAML_LOGINS: + db.session.commit() + return False + db.session.add(SamlLogin(request_id, idp_identity_name)) + db.session.commit() + return True + + +class OutstandingSamlRequests(collections.abc.Mapping): + """ + The AuthnRequests sent to an IdP that it can still answer, as pysaml2 expects them in parse_authn_request_response + + Looked up in the database, a response can arrive at another process of the server than the one that sent the + request. pysaml2 refuses a response to a request that isn't one of them, also if the SP configuration has + allow_unsolicited; finish_saml_login refuses one without a request. + """ + + def __init__(self, idp_identity_name): + self.idp_identity_name = idp_identity_name + # pysaml2 looks up the same request several times + self._found = {} + + def __getitem__(self, request_id): + if request_id is None: + raise KeyError(request_id) + if request_id not in self._found: + self._found[request_id] = _saml_requests(self.idp_identity_name, request_id).first() is not None + if not self._found[request_id]: + raise KeyError(request_id) + # where the login started, pysaml2 keeps it as came_from; unused + return "/" + + def __iter__(self): + # pysaml2 only logs the keys + return iter(()) + + def __len__(self): + return 0 + + def __bool__(self): + # pysaml2 ignores an empty mapping + return True + + +def saml_attributes(authn_response): + """ + The attributes of a SAML response as {name: [values]}, by their names in the response and by the friendly names + of pysaml2's attribute maps + """ + attributes = {} + assertion = getattr(authn_response, "assertion", None) + for statement in (assertion.attribute_statement if assertion is not None else []): + for attribute in statement.attribute: + values = [value.text for value in attribute.attribute_value if value.text] + for name in (attribute.name, attribute.friendly_name): + if name: + attributes.setdefault(name, values) + for name, values in (authn_response.ava or {}).items(): + attributes[name] = [value for value in values if isinstance(value, str)] + return attributes + + +def saml_identity(authn_response, idp_data, attributes): + """ + (issuer, subject) of a SAML response, the identity a user is linked to; None if the response has no persistent + identifier + + The issuer is the entity id of the IdP that signed the assertion: the metadata of one configured IdP can hold + several, e.g. of a federation. The issuer of the response around it may be unsigned. An unsigned assertion + (want_assertion_signed: false) counts only if the signed response has the same issuer: pysaml2 checks the + signature of an assertion or response that has one, so only a signature ties the issuer to its key. The subject + is the value of the attribute idp_data['user_id_attribute'] in attributes, see saml_attributes, or the NameID if + that isn't set and the NameID is persistent. A transient NameID changes with every login, an email address can + be given to someone else. Logs why a response has no identity. + """ + assertion = getattr(authn_response, "assertion", None) + issuer = assertion.issuer.text if assertion is not None and assertion.issuer is not None else None + if issuer and getattr(assertion, "signature", None) is None: + response = getattr(authn_response, "response", None) + if getattr(response, "signature", None) is None or authn_response.issuer() != issuer: + logging.warning("Refused a SAML response with an unsigned assertion of %r: the response around it isn't " + "signed by the same IdP", issuer) + return None + attribute = idp_data.get('user_id_attribute') + if attribute: + values = attributes.get(attribute, []) + # of several values none is the identifier + subject = values[0] if len(values) == 1 else None else: - user.authentication_backend = authentication_backend - user.hash_password(token) - result = fm.modify_user(user, action="update_idp_user") - return result + name_id = authn_response.name_id + subject = name_id.text if name_id is not None and name_id.format == NAMEID_FORMAT_PERSISTENT else None + if not issuer: + logging.warning("Refused a SAML response without issuer") + return None + if not subject: + logging.warning("Refused a SAML response of %r without a persistent identifier: set 'user_id_attribute' of " + "the IdP to a single valued attribute that identifies a user", issuer) + return None + if len(issuer) > 255 or len(subject) > 255: + logging.warning("Refused a SAML response of %r: its issuer or identifier is longer than 255 characters", issuer) + return None + return issuer, subject + + +class SamlLoginRefused(Exception): + """A SAML login that can't log in, the message tells the user why""" + + +def user_of_email(emailid): + """ + The user of emailid, whatever its case, None if there is none; an exact match comes first + + Raises SamlLoginRefused if several accounts have emailid in different case: which one is meant is unknown. + """ + user = User.query.filter_by(emailid=str(emailid)).first() + if user is None: + users = User.query.filter(sqlalchemy.func.lower(User.emailid) == str(emailid).lower()).limit(2).all() + if len(users) > 1: + raise SamlLoginRefused("Several accounts of this server have your email address in different upper and " + "lower case. Ask its administrator for help.") + user = users[0] if users else None + return user + + +def _new_saml_username(emailid, username): + """ + The username of a new account for a SAML login: the one that the IdP sent, else, also if that is taken or + can't be used, the start of the email; raises SamlLoginRefused if register_user would refuse both + """ + candidates = [name.strip() for name in (username, emailid.split("@")[0]) if name and name.strip()] + for candidate in candidates: + if len(candidate) <= 255 and "@" not in candidate \ + and User.query.filter_by(username=candidate).first() is None: + return candidate + raise SamlLoginRefused(f"The usernames {' and '.join(repr(name) for name in candidates)} that your identity " + "provider sent are taken or can't be used for an account of this server. Ask its " + "administrator for help.") + + +def check_saml_login(issuer, subject, emailid, username): + """ + Checks the email and username that the IdP sent with the identity (issuer, subject); returns them as + finish_saml_login stores them, raises SamlLoginRefused if the login can't log in + + A linked identity needs neither. Else the email must be valid: it may belong to an existing account, which must + be linked, see idp_login. Else a new account is created, its username is checked like by register_user. + """ + if IdpIdentity.query.filter_by(issuer=str(issuer), subject=str(subject)).first() is not None: + return None, None + if not isinstance(emailid, str) or len(emailid) > 255: + raise SamlLoginRefused("Your identity provider sent no valid email address.") + try: + email_validator.validate_email(emailid, check_deliverability=False) + except email_validator.EmailNotValidError: + raise SamlLoginRefused("Your identity provider sent no valid email address.") + if user_of_email(emailid) is not None: + return emailid, None + return emailid, _new_saml_username(emailid, username) + + +def _code_hash(code): + # the code is random and long, a plain hash is enough to not store it + return hashlib.sha256(code.encode("utf-8")).hexdigest() + + +def finish_saml_login(request_id, idp_identity_name, issuer, subject, emailid, username): + """ + Stores the identity that the IdP asserts in its response to the login request_id, with emailid and username + checked by check_saml_login; returns a one-time code that msui exchanges for a login token by idp_login, None if + the IdP can't answer request_id (any more) + + Only one response per request: a second one, e.g. the first sent again, is refused. + """ + if request_id is None: + return None + code = secrets.token_urlsafe(32) + updated = _saml_requests(idp_identity_name, request_id).update({ + SamlLogin.code_hash: _code_hash(code), + SamlLogin.issuer: str(issuer), + SamlLogin.subject: str(subject), + SamlLogin.emailid: None if emailid is None else str(emailid), + SamlLogin.username: None if username is None else str(username), + SamlLogin.created_on: _now(), + }, synchronize_session=False) + db.session.commit() + return code if updated == 1 else None + + +_SamlLoginResult = collections.namedtuple("_SamlLoginResult", "id idp_identity_name issuer subject emailid username") + + +def _saml_login_of_code(code): + # the finished SAML login of code, a copy: the row is gone once the code is used + if not isinstance(code, str) or not code: + return None + saml_login = SamlLogin.query.filter( + SamlLogin.code_hash == _code_hash(code), + SamlLogin.created_on >= _now() - datetime.timedelta(seconds=IDP_LOGIN_CODE_LIFETIME)).first() + if saml_login is None: + return None + return _SamlLoginResult(saml_login.id, saml_login.idp_identity_name, saml_login.issuer, saml_login.subject, + saml_login.emailid, saml_login.username) + + +def _use_saml_login(saml_login, user, link=False, first_identity=False): + """ + Uses up the code of saml_login: stores user, if it is new, and links the identity of saml_login to it, if link + + All in one transaction, so nothing changes if it fails: another request used the code first, linked the identity, + or created the account meanwhile; or first_identity and the user has an identity meanwhile. Returns whether it + succeeded. + """ + # the checks below read the state of now, not of the reads before + db.session.commit() + try: + if SamlLogin.query.filter_by(id=saml_login.id).delete(synchronize_session=False) != 1: + db.session.rollback() + return False + if user.id is None: + db.session.add(user) + db.session.flush() + elif first_identity: + # waits for another request that links an identity to user at the same time + User.query.filter_by(id=user.id).with_for_update().one() + if IdpIdentity.query.filter_by(u_id=user.id).count() > 0: + db.session.rollback() + return False + if link: + db.session.add(IdpIdentity(user.id, saml_login.issuer, saml_login.subject)) + db.session.commit() + except sqlalchemy.exc.IntegrityError: + db.session.rollback() + return False + return True + + +def _refuse_unconfirmed(user): + # like /token, before the code is used: the user can confirm the email and try the same code again + if current_app.config['MAIL_ENABLED'] and not user.confirmed: + raise SamlLoginRefused("Your account isn't confirmed yet. Confirm your email address, then enter the code " + "again.") + + +def idp_login(code): + """ + The user that the one-time code of a SAML login logs in + + Returns (user, None) and makes the code invalid. Returns (None, emailid) if the identity isn't linked to a user + and its email belongs to an existing account: the user must log in to that account to link it, see + link_idp_identity, the code stays valid for that. Returns (None, None) if the code is invalid or expired, or the + login can't log in, see check_saml_login. Raises SamlLoginRefused, with a message for the user, if the account + isn't confirmed (with MAIL_ENABLED) or its email is ambiguous; the code stays valid then. + + A new identity gets a new account. A SAML login never changes the password of an account. + """ + saml_login = _saml_login_of_code(code) + if saml_login is None: + return None, None + identity = IdpIdentity.query.filter_by(issuer=saml_login.issuer, subject=saml_login.subject).first() + if identity is not None: + user = identity.user + _refuse_unconfirmed(user) + return (user, None) if _use_saml_login(saml_login, user) else (None, None) + if not saml_login.emailid: + logging.warning("The SAML login of %s has no email, can't create an account", saml_login.idp_identity_name) + return None, None + user = user_of_email(saml_login.emailid) + if user is not None: + # Accounts of older versions logged in by the email that the IdP sent. One that logged in by this IdP before + # and has no identity yet is linked on its first login, if the response comes from the IdP the login request + # went to: the metadata of a federation lets any of its IdPs answer and assert the email. Any other account + # must be linked by its owner. + if user.authentication_backend != saml_login.idp_identity_name or user.idp_identities \ + or saml_login.issuer != get_idp_entity_id(saml_login.idp_identity_name): + return None, user.emailid + _refuse_unconfirmed(user) + linked = _use_saml_login(saml_login, user, link=True, first_identity=True) + return (user, None) if linked else (None, None) + try: + username = _new_saml_username(saml_login.emailid, saml_login.username) + except SamlLoginRefused as ex: + # an account with the username was created since the IdP answered + logging.warning("Refused the SAML login of %s: %s", saml_login.idp_identity_name, ex) + return None, None + # nobody knows the password, the owner can set one by a password reset + user = User(saml_login.emailid, username, password=secrets.token_urlsafe(32), confirmed=True, + confirmed_on=_now() + datetime.timedelta(seconds=1), + authentication_backend=saml_login.idp_identity_name) + return (user, None) if _use_saml_login(saml_login, user, link=True) else (None, None) + + +def link_idp_identity(user, code): + """ + Links the identity of a SAML login to user, if idp_login refused its code because its email is the one of user; + returns whether it was linked + + user is logged in, so whoever links knows the password of the account and can log in at the IdP. + """ + saml_login = _saml_login_of_code(code) + if saml_login is None or not saml_login.emailid or saml_login.emailid.lower() != user.emailid.lower(): + return False + return _use_saml_login(saml_login, user, link=True) diff --git a/mslib/mscolab/blueprints/auth/__init__.py b/mslib/mscolab/blueprints/auth/__init__.py index e0b390cb2..280a1abd0 100644 --- a/mslib/mscolab/blueprints/auth/__init__.py +++ b/mslib/mscolab/blueprints/auth/__init__.py @@ -26,25 +26,29 @@ """ import datetime +import functools import json import logging -import secrets -from flask import Blueprint, request, url_for, render_template, jsonify, flash, redirect, current_app +from flask import Blueprint, request, url_for, render_template, jsonify, flash, redirect, current_app, abort, g from flask.wrappers import Response from saml2 import BINDING_HTTP_REDIRECT, BINDING_HTTP_POST from saml2.metadata import create_metadata_string +from saml2.saml import NAMEID_FORMAT_PERSISTENT -from mslib.mscolab.auth import check_login, register_user, generate_confirmation_token, create_or_update_idp_user, \ - get_idp_entity_id, confirm_token, generate_password_reset_token, confirm_password_reset_token, public_url_for, \ - EMAIL_CONFIRMATION, IDP_LOGIN -from mslib.mscolab.auth import optional_auth, request_token +from mslib.mscolab.auth import check_login, register_user, generate_confirmation_token, get_idp_entity_id, \ + confirm_token, generate_password_reset_token, confirm_password_reset_token, public_url_for, EMAIL_CONFIRMATION +from mslib.mscolab.auth import optional_auth, request_token, verify_user +from mslib.mscolab.auth import OutstandingSamlRequests, SamlLoginRefused, check_saml_login, find_idp_config, \ + finish_saml_login, idp_login, link_idp_identity, remember_saml_request, saml_attributes, saml_identity from mslib.mscolab.conf import setup_saml2_backend from mslib.mscolab.forms import ResendConfirmationForm, ResetPasswordForm, ResetRequestForm from mslib.mscolab.models import User from mslib.mscolab.api import endpoints from mslib.mscolab.api.attachments import normalized_extensions from mslib.mscolab.api.schemas import ( + IdpLinkRequest, + IdpLinkResponse, IdpLoginAuthRequest, IdpLoginAuthResponse, IdpUserInfo, @@ -139,151 +143,174 @@ def user_register_handler(): return jsonify(response.to_dict()), status_code +def saml_required(func): + """The SAML routes exist only with USE_SAML2""" + @functools.wraps(func) + def wrapper(*args, **kwargs): + if not current_app.config.get("USE_SAML2"): + abort(404) + return func(*args, **kwargs) + return wrapper + + def init_saml(state): - """Runs once when blueprint is registered.""" - app = state.app + """ + Adds the assertion consumer services of the configured IdPs to the app, if it uses SAML + Their paths come from the configuration of each IdP, so they are added to the app and not to the blueprint, which + can't change once it was registered with an app. + """ + app = state.app if not app.config.get("USE_SAML2"): return - - setup_saml2_backend() - register_saml_routes() - - -def register_saml_routes(): - """All SAML routes are registered here safely.""" - - def create_acs_post_handler(idp_config): - """ - Create acs_post_handler function for the given idp_config. - """ - - def acs_post_handler(): - """ - Function to handle SAML authentication response. - """ - try: - outstanding_queries = {} - binding = BINDING_HTTP_POST - authn_response = idp_config['idp_data']['saml2client'].parse_authn_request_response( - request.form["SAMLResponse"], binding, outstanding=outstanding_queries - ) - - try: - email = authn_response.ava["email"][0] - username = authn_response.ava["givenName"][0] - token = generate_confirmation_token(email, IDP_LOGIN) - except (NameError, AttributeError, KeyError): - try: - # Initialize an empty dictionary to store attribute values - attributes = {} - - # Loop through attribute statements - for attribute_statement in authn_response.assertion.attribute_statement: - for attribute in attribute_statement.attribute: - attribute_name = attribute.name - attribute_value = \ - attribute.attribute_value[0].text if attribute.attribute_value else None - attributes[attribute_name] = attribute_value - - # Extract the email and givenname attributes - email = attributes["email"] - username = attributes["givenName"] - token = generate_confirmation_token(email, IDP_LOGIN) - except (NameError, AttributeError, KeyError): - return render_template('auth/errors/403.html'), 403 - - if email is not None and username is not None: - idp_identity_name = idp_config['idp_identity_name'] - idp_user_db_state = create_or_update_idp_user(email, - username, token, idp_identity_name) - if idp_user_db_state: - return render_template('auth/idp/idp_login_success.html', token=token), 200 - return render_template('auth/errors/500.html'), 500 - return render_template('auth/errors/500.html'), 500 - except (NameError, AttributeError, KeyError): - return render_template('auth/errors/403.html'), 403 - - return acs_post_handler - - # Implementation for handling configured SAML assertion consumer endpoints for idp_config in setup_saml2_backend.CONFIGURED_IDPS: try: - for assertion_consumer_endpoint in idp_config['idp_data']['assertion_consumer_endpoints']: - # Dynamically add the route for the current endpoint - AUTH_BP.add_url_rule(f'/{assertion_consumer_endpoint}/', assertion_consumer_endpoint, - create_acs_post_handler(idp_config), methods=['POST']) - except (NameError, AttributeError, KeyError) as ex: - logging.warning("USE_SAML2 is %s, Failure is: %s", current_app.config['USE_SAML2'], ex) - - @AUTH_BP.route('/available_idps/', methods=['GET']) - def available_idps(): - """ - This function checks if IDP (Identity Provider) is enabled in the mscolab_settings module. - If IDP is enabled, it retrieves the configured IDPs from setup_saml2_backend.CONFIGURED_IDPS - and renders the 'idp/available_idps.html' template with the list of configured IDPs. - """ - configured_idps = setup_saml2_backend.CONFIGURED_IDPS - return render_template('auth/idp/available_idps.html', configured_idps=configured_idps), 200 - - @AUTH_BP.route("/idp_login/", methods=['POST']) - def idp_login(): - """Handle the login process for the user by selected IDP""" - selected_idp = request.form.get('selectedIdentityProvider') - sp_config = None - for config_ in setup_saml2_backend.CONFIGURED_IDPS: - if selected_idp == config_['idp_identity_name']: - sp_config = config_['idp_data']['saml2client'] - break - + endpoints_ = idp_config['idp_data']['assertion_consumer_endpoints'] + except KeyError as ex: + logging.warning("The IdP %s has no assertion consumer service: %s", idp_config.get('idp_identity_name'), ex) + continue + if not idp_config['idp_data'].get('user_id_attribute'): + name_id_formats = idp_config['idp_data']['saml2client'].config.getattr("name_id_format", "sp") or [] + if NAMEID_FORMAT_PERSISTENT not in ([name_id_formats] if isinstance(name_id_formats, str) + else name_id_formats): + logging.warning("The IdP %s has no 'user_id_attribute' and the SP asks for no persistent NameID: " + "every SAML login by it is refused. Set 'user_id_attribute' in " + "setup_saml2_backend.py, see docs/sso_via_saml_mscolab.rst", + idp_config['idp_identity_name']) + for index, path in enumerate(endpoints_): + # as configured, also with a trailing slash: the metadata tells the IdP this URL + app.add_url_rule(f"/{path.lstrip('/')}", f"saml_acs_{idp_config['idp_identity_name']}_{index}", + _acs_post_handler(idp_config), methods=['POST'], strict_slashes=False) + + +def _first(values): + return values[0] if values else None + + +def _acs_post_handler(idp_config): + """The handler of the SAML responses of the IdP idp_config""" + idp_identity_name = idp_config['idp_identity_name'] + + def acs_post_handler(): try: - _, response_binding = sp_config.config.getattr("endpoints", "sp")[ - "assertion_consumer_service" - ][0] - entity_id = get_idp_entity_id(selected_idp) - _, binding, http_args = sp_config.prepare_for_negotiated_authenticate( - entityid=entity_id, - response_binding=response_binding, - ) - if binding == BINDING_HTTP_REDIRECT: - headers = dict(http_args["headers"]) - return redirect(str(headers["Location"]), code=303) - return Response(http_args["data"], headers=http_args["headers"]) - except (NameError, AttributeError): + authn_response = idp_config['idp_data']['saml2client'].parse_authn_request_response( + request.form["SAMLResponse"], BINDING_HTTP_POST, outstanding=OutstandingSamlRequests(idp_identity_name)) + except Exception as ex: + # pysaml2 raises many kinds of exceptions for an invalid, unsigned, expired or unsolicited response + logging.warning("Refused a SAML response of %s: %s: %s", idp_identity_name, type(ex).__name__, ex) + return render_template('auth/errors/403.html'), 403 + if authn_response is None: + return render_template('auth/errors/403.html'), 403 + attributes = saml_attributes(authn_response) + identity = saml_identity(authn_response, idp_config['idp_data'], attributes) + if identity is None: + # saml_identity logged why return render_template('auth/errors/403.html'), 403 - - @AUTH_BP.route(f"/{endpoints.IDP_LOGIN_AUTH}/", methods=['POST']) - def idp_login_auth(): - """Handle the SAML authentication validation of client application.""" try: - req = IdpLoginAuthRequest.from_json_data(request.get_json()) - email = confirm_token(req.token, IDP_LOGIN, expiration=1200) - if email: - user = check_login(email, req.token) - if user: - fm = current_app.extensions['fm'] - random_token = secrets.token_hex(16) - user.hash_password(random_token) - fm.modify_user(user, action="update_idp_user") - user_info = IdpUserInfo(username=user.username, id=user.id, emailid=user.emailid) - response = IdpLoginAuthResponse(success=True, token=random_token, user=user_info) - return response.to_text() - return jsonify(IdpLoginAuthResponse(success=False).to_dict()), 401 - return jsonify(IdpLoginAuthResponse(success=False).to_dict()), 401 - except TypeError: - return jsonify(IdpLoginAuthResponse(success=False).to_dict()), 401 - - @AUTH_BP.route("/metadata/", methods=['GET']) - def metadata(idp_identity_name): - """Return the SAML metadata XML for the requested IDP""" - for config_ in setup_saml2_backend.CONFIGURED_IDPS: - if idp_identity_name == config_['idp_identity_name']: - sp_config = config_['idp_data']['saml2client'] - metadata_string = create_metadata_string( - None, sp_config.config, 4, None, None, None, None, None - ).decode("utf-8") - return Response(metadata_string, mimetype="text/xml") + email, username = check_saml_login(*identity, _first(attributes.get("email") or attributes.get("mail")), + _first(attributes.get("givenName") or attributes.get("displayName"))) + except SamlLoginRefused as ex: + logging.warning("Refused a SAML response of %s: %s", idp_identity_name, ex) + return render_template('auth/errors/403.html', message=str(ex)), 403 + code = finish_saml_login(authn_response.in_response_to, idp_identity_name, *identity, email, username) + if code is None: + # no login of this server waits for this response: it was sent again, or too late + logging.warning("Refused a SAML response of %s that answers no open login", idp_identity_name) + return render_template('auth/errors/403.html'), 403 + return render_template('auth/idp/idp_login_success.html', code=code), 200 + + return acs_post_handler + + +@AUTH_BP.route('/available_idps/', methods=['GET']) +@saml_required +@optional_auth +def available_idps(): + """The page to choose one of the IdPs of setup_saml2_backend.CONFIGURED_IDPS for a login""" + configured_idps = setup_saml2_backend.CONFIGURED_IDPS + return render_template('auth/idp/available_idps.html', configured_idps=configured_idps), 200 + + +@AUTH_BP.route("/idp_login/", methods=['POST']) +@saml_required +@optional_auth +def idp_login_handler(): + """Starts a login by the selected IdP: sends the browser there with an AuthnRequest""" + selected_idp = request.form.get('selectedIdentityProvider') + idp_config = find_idp_config(selected_idp) + if idp_config is None: + return render_template('auth/errors/404.html'), 404 + try: + sp_config = idp_config['idp_data']['saml2client'] + _, response_binding = sp_config.config.getattr("endpoints", "sp")["assertion_consumer_service"][0] + entity_id = get_idp_entity_id(selected_idp) + request_id, binding, http_args = sp_config.prepare_for_negotiated_authenticate( + entityid=entity_id, + response_binding=response_binding, + ) + except (NameError, AttributeError, KeyError, IndexError) as ex: + logging.warning("Can't create a SAML request for %s: %s", selected_idp, ex) + return render_template('auth/errors/403.html'), 403 + # the response must answer this request + if not remember_saml_request(request_id, selected_idp): + logging.warning("Refused a SAML login by %s: too many logins are unfinished", selected_idp) + return render_template('auth/errors/503.html'), 503 + if binding == BINDING_HTTP_REDIRECT: + headers = dict(http_args["headers"]) + return redirect(str(headers["Location"]), code=303) + return Response(http_args["data"], headers=http_args["headers"]) + + +@AUTH_BP.route(f"/{endpoints.IDP_LOGIN_AUTH}/", methods=['POST']) +@saml_required +@optional_auth +def idp_login_auth(): + """ + Exchanges the one-time code of a SAML login for a login token + + 409 if the SAML identity must first be linked to the existing account of its email, see idp_link; 403 with a + message if the login is refused, e.g. for an account that isn't confirmed (like /token). Like /token behind the + HTTP basic authentication of the server, if it has one. + """ + data = request.get_json(silent=True) + req = IdpLoginAuthRequest.from_json_data(data if isinstance(data, dict) else {}) + try: + user, emailid = idp_login(req.token) + except SamlLoginRefused as ex: + return jsonify(IdpLoginAuthResponse(success=False, message=str(ex)).to_dict()), 403 + if user is not None: + user_info = IdpUserInfo(username=user.username, id=user.id, emailid=user.emailid, fullname=user.fullname) + return IdpLoginAuthResponse(success=True, token=user.generate_auth_token(), user=user_info).to_text() + if emailid is not None: + return jsonify(IdpLoginAuthResponse(success=False, link_required=True, emailid=emailid).to_dict()), 409 + return jsonify(IdpLoginAuthResponse(success=False).to_dict()), 401 + + +@AUTH_BP.route(f"/{endpoints.IDP_LINK}", methods=['POST']) +@saml_required +@optional_auth +@verify_user +def idp_link(): + """ + Links the SAML identity of a one-time code, that idp_login_auth refused with link_required, to the logged in user + + The code must be one for the email of the user. + """ + req = IdpLinkRequest.from_form(request.form) + return jsonify(IdpLinkResponse(success=link_idp_identity(g.user, req.code)).to_dict()) + + +@AUTH_BP.route("/metadata/", methods=['GET']) +@saml_required +def metadata(idp_identity_name): + """Return the SAML metadata XML for the requested IDP""" + idp_config = find_idp_config(idp_identity_name) + if idp_config is None: return render_template('auth/errors/404.html'), 404 + metadata_string = create_metadata_string( + None, idp_config['idp_data']['saml2client'].config, 4, None, None, None, None, None + ).decode("utf-8") + return Response(metadata_string, mimetype="text/xml") AUTH_BP.record_once(init_saml) diff --git a/mslib/mscolab/blueprints/auth/templates/auth/errors/403.html b/mslib/mscolab/blueprints/auth/templates/auth/errors/403.html index 7dca5ccd7..057e6a042 100644 --- a/mslib/mscolab/blueprints/auth/templates/auth/errors/403.html +++ b/mslib/mscolab/blueprints/auth/templates/auth/errors/403.html @@ -2,4 +2,7 @@

403 - Forbidden

You don't have permission to access this page

+ {% if message %} +

{{ message }}

+ {% endif %} diff --git a/mslib/mscolab/blueprints/auth/templates/auth/errors/503.html b/mslib/mscolab/blueprints/auth/templates/auth/errors/503.html new file mode 100644 index 000000000..73fac063b --- /dev/null +++ b/mslib/mscolab/blueprints/auth/templates/auth/errors/503.html @@ -0,0 +1,5 @@ +
+

503 - Service Unavailable

+
+

The server can't take more logins right now. Please try again in a few minutes.

+
diff --git a/mslib/mscolab/blueprints/auth/templates/auth/idp/available_idps.html b/mslib/mscolab/blueprints/auth/templates/auth/idp/available_idps.html index e6ad7eab4..ce324083f 100644 --- a/mslib/mscolab/blueprints/auth/templates/auth/idp/available_idps.html +++ b/mslib/mscolab/blueprints/auth/templates/auth/idp/available_idps.html @@ -48,24 +48,15 @@

Choose Identity Provider

-
+
    {% for idp in configured_idps %}
  • - +
    • {% endfor %} -
-
-
{% endblock %} diff --git a/mslib/mscolab/blueprints/auth/templates/auth/idp/idp_login_success.html b/mslib/mscolab/blueprints/auth/templates/auth/idp/idp_login_success.html index e81d8ffbb..01f6d4364 100644 --- a/mslib/mscolab/blueprints/auth/templates/auth/idp/idp_login_success.html +++ b/mslib/mscolab/blueprints/auth/templates/auth/idp/idp_login_success.html @@ -9,34 +9,26 @@

-

Congratulations! You have successfully logged in to the mscolab server using Identity Provider.

-

Please proceed to log in using the user interface by below token.

-

Token : {{token}} +

You have logged in to the MSColab server by your identity provider.

+

Enter this code in the user interface within 20 minutes to finish the login. It works once.

+

Code: {{ code }}
- +

- +

If an MSColab account with your email exists already, the user interface asks for its password once, to link + this login to the account.

diff --git a/mslib/mscolab/conf.py b/mslib/mscolab/conf.py index 30cd5d21c..685fa2ec6 100644 --- a/mslib/mscolab/conf.py +++ b/mslib/mscolab/conf.py @@ -172,6 +172,9 @@ class setup_saml2_backend: 'idp_identity_name': 'localhost_test_idp', 'idp_data': { 'idp_name': 'Testing Identity Provider', + # the attribute that identifies a user, it must not change and not be given to someone else; + # without it, the IdP must send a persistent NameID + 'user_id_attribute': 'eduPersonPrincipalName', } }, @@ -179,6 +182,7 @@ class setup_saml2_backend: # 'idp_identity_name': 'idp2', # 'idp_data': { # 'idp_name': '2nd Identity Provider', + # 'user_id_attribute': 'eduPersonPrincipalName', # } # }, ] diff --git a/mslib/mscolab/file_manager.py b/mslib/mscolab/file_manager.py index 37abaf12d..860296086 100644 --- a/mslib/mscolab/file_manager.py +++ b/mslib/mscolab/file_manager.py @@ -304,13 +304,6 @@ def modify_user(self, user, attribute=None, value=None, action=None): # on delete we return successful deleted if user_query is None: return True - elif action == "update_idp_user": - user_query = User.query.filter_by(emailid=str(user.emailid)).first() - if user_query is not None: - db.session.add(user) - db.session.commit() - else: - return False # This is the default, when we not have a special action user_query = User.query.filter_by(id=user.id).first() if user_query is None: diff --git a/mslib/mscolab/migrations/schema_repair.py b/mslib/mscolab/migrations/schema_repair.py index 16601948c..b8af79480 100644 --- a/mslib/mscolab/migrations/schema_repair.py +++ b/mslib/mscolab/migrations/schema_repair.py @@ -135,6 +135,48 @@ def add_token_nonce(op, conn): batch_op.alter_column('token_nonce', existing_type=sa.String(length=32), nullable=False) +def has_table(conn, table): + """True if table exists in the database of conn""" + return table in sa.inspect(conn).get_table_names() + + +def add_saml_tables(op, conn): + """ + Adds the tables idp_identities, the SAML identities linked to a user, and saml_logins, the SAML logins in progress, + if they are missing + + op is an alembic Operations object for conn. + """ + if not has_table(conn, 'idp_identities'): + op.create_table( + 'idp_identities', + sa.Column('id', sa.Integer(), autoincrement=True, nullable=False), + sa.Column('u_id', sa.Integer(), nullable=False), + sa.Column('issuer', sa.String(length=255), nullable=False), + sa.Column('subject', sa.String(length=255), nullable=False), + sa.Column('created_on', sa.DateTime(), nullable=False), + sa.ForeignKeyConstraint(['u_id'], ['users.id'], name=op.f('fk_idp_identities_u_id_users')), + sa.PrimaryKeyConstraint('id', name=op.f('pk_idp_identities')), + sa.UniqueConstraint('issuer', 'subject', name=op.f('uq_idp_identities_issuer')), + ) + if not has_table(conn, 'saml_logins'): + op.create_table( + 'saml_logins', + sa.Column('id', sa.Integer(), autoincrement=True, nullable=False), + sa.Column('request_id', sa.String(length=255), nullable=False), + sa.Column('idp_identity_name', sa.String(length=255), nullable=False), + sa.Column('created_on', sa.DateTime(), nullable=False), + sa.Column('code_hash', sa.String(length=64), nullable=True), + sa.Column('issuer', sa.String(length=255), nullable=True), + sa.Column('subject', sa.String(length=255), nullable=True), + sa.Column('emailid', sa.String(length=255), nullable=True), + sa.Column('username', sa.String(length=255), nullable=True), + sa.PrimaryKeyConstraint('id', name=op.f('pk_saml_logins')), + sa.UniqueConstraint('request_id', name=op.f('uq_saml_logins_request_id')), + sa.UniqueConstraint('code_hash', name=op.f('uq_saml_logins_code_hash')), + ) + + _MISSING_USER = "{column} IS NOT NULL AND {column} NOT IN (SELECT id FROM users)" _MISSING_OPERATION = "op_id IS NOT NULL AND op_id NOT IN (SELECT id FROM operations)" diff --git a/mslib/mscolab/migrations/versions/3a1c5e2b9d47_to_version_12_0_0.py b/mslib/mscolab/migrations/versions/3a1c5e2b9d47_to_version_12_0_0.py index 1f4105084..787f96e2b 100644 --- a/mslib/mscolab/migrations/versions/3a1c5e2b9d47_to_version_12_0_0.py +++ b/mslib/mscolab/migrations/versions/3a1c5e2b9d47_to_version_12_0_0.py @@ -10,9 +10,10 @@ import sqlalchemy as sa from alembic import op -from mslib.mscolab.migrations.schema_repair import (add_fullname, add_token_nonce, batch_alter_table, - hand_over_operations_without_creator, has_column, - remove_dangling_references, use_autoincrement_ids) +from mslib.mscolab.migrations.schema_repair import (add_fullname, add_saml_tables, add_token_nonce, + batch_alter_table, hand_over_operations_without_creator, + has_column, has_table, remove_dangling_references, + use_autoincrement_ids) # revision identifiers, used by Alembic. @@ -67,15 +68,23 @@ def upgrade(): if not has_column(conn, 'users', 'token_nonce'): add_token_nonce(op, conn) + # SAML identities linked to a user, and the SAML logins in progress + add_saml_tables(op, conn) + # SQLite: the id of a deleted user or operation is never handed out again use_autoincrement_ids(op, conn) def downgrade(): # AUTOINCREMENT of users and operations on SQLite stays, an earlier version works with it - # missing in a database that ran an earlier development state of this migration - if has_column(op.get_bind(), 'users', 'token_nonce'): - with batch_alter_table(op, op.get_bind(), 'users') as batch_op: + # missing, like token_nonce, in a database that ran an earlier development state of this migration + conn = op.get_bind() + for table in ('saml_logins', 'idp_identities'): + if has_table(conn, table): + op.drop_table(table) + + if has_column(conn, 'users', 'token_nonce'): + with batch_alter_table(op, conn, 'users') as batch_op: batch_op.drop_column('token_nonce') with op.batch_alter_table('permissions', schema=None) as batch_op: diff --git a/mslib/mscolab/models.py b/mslib/mscolab/models.py index b7465a67c..d9330b2cf 100644 --- a/mslib/mscolab/models.py +++ b/mslib/mscolab/models.py @@ -88,6 +88,9 @@ class User(db.Model): confirmed = db.Column(db.Boolean, nullable=False, default=False) confirmed_on = db.Column(AwareDateTime, nullable=True) permissions = db.relationship('Permission', cascade='all,delete,delete-orphan', backref='user') + idp_identities = db.relationship('IdpIdentity', cascade='all,delete,delete-orphan', backref='user') + # the IdP of the first SAML login of the account, 'local' for a registered one; only informative, a SAML login is + # accepted by IdpIdentity. Accounts of older versions have no IdpIdentity, see mslib.mscolab.auth.idp_login authentication_backend = db.Column(db.String(255), nullable=False, default='local') fullname = db.Column(db.String(255), nullable=True) # random value in every login token of the user; a new one makes all earlier tokens invalid, see revoke_tokens. @@ -229,10 +232,19 @@ def reset_password(self, password_hash, password): For a password reset link, which works once: of two requests with the same link at the same time only the first changes the password, the stored hash no longer matches for the second. Commits the session. + + Also unlinks the SAML identities of the user: whoever resets the password owns the email address, an IdP that + lets its users choose their email could have linked another person's identity to the account before. A SAML + user links again on the next SAML login, with the new password. The account counts as a password account + from then on (authentication_backend 'local'): the first login of an account of an older version links it + without the password, see auth.idp_login. """ updated = User.query.filter_by(id=self.id, password=password_hash).update( - {User.password: PH.hash(password), User.token_nonce: secrets.token_hex(16), User.confirmed: True}, + {User.password: PH.hash(password), User.token_nonce: secrets.token_hex(16), User.confirmed: True, + User.authentication_backend: "local"}, synchronize_session=False) + if updated == 1: + IdpIdentity.query.filter_by(u_id=self.id).delete(synchronize_session=False) db.session.commit() db.session.refresh(self) if updated == 1 and self.confirmed_on is None: @@ -242,6 +254,61 @@ def reset_password(self, password_hash, password): return updated == 1 +class IdpIdentity(db.Model): + """ + A SAML identity that may log in as its user: the entity id of the IdP that asserts it, and the subject, the value + of the configured identifier attribute or the persistent NameID + + A user can have several, e.g. from different IdPs. An identity belongs to one user. + """ + + __tablename__ = 'idp_identities' + __table_args__ = (db.UniqueConstraint('issuer', 'subject'),) + id = db.Column(db.Integer, primary_key=True, autoincrement=True) # noqa: A003 + u_id = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=False) + issuer = db.Column(db.String(255), nullable=False) + subject = db.Column(db.String(255), nullable=False) + created_on = db.Column(AwareDateTime, nullable=False) + + def __init__(self, u_id, issuer, subject): + self.u_id = int(u_id) + self.issuer = str(issuer) + self.subject = str(subject) + self.created_on = datetime.datetime.now(tz=datetime.timezone.utc) + + def __repr__(self): + return f'' + + +class SamlLogin(db.Model): + """ + A SAML login in progress + + Created with the id of the AuthnRequest sent to the IdP. The response of the IdP must answer it, and only once: + it adds the identity it asserts and the hash of a one-time code, which msui exchanges for a login token. + """ + + __tablename__ = 'saml_logins' + id = db.Column(db.Integer, primary_key=True, autoincrement=True) # noqa: A003 + request_id = db.Column(db.String(255), unique=True, nullable=False) + idp_identity_name = db.Column(db.String(255), nullable=False) + # when the request was sent, and when the response arrived + created_on = db.Column(AwareDateTime, nullable=False) + code_hash = db.Column(db.String(64), unique=True, nullable=True) + issuer = db.Column(db.String(255), nullable=True) + subject = db.Column(db.String(255), nullable=True) + emailid = db.Column(db.String(255), nullable=True) + username = db.Column(db.String(255), nullable=True) + + def __init__(self, request_id, idp_identity_name): + self.request_id = str(request_id) + self.idp_identity_name = str(idp_identity_name) + self.created_on = datetime.datetime.now(tz=datetime.timezone.utc) + + def __repr__(self): + return f'' + + class Permission(db.Model): __tablename__ = 'permissions' diff --git a/mslib/mscolab/mscolab.py b/mslib/mscolab/mscolab.py index 2909e4dcb..e209ff26c 100644 --- a/mslib/mscolab/mscolab.py +++ b/mslib/mscolab/mscolab.py @@ -265,7 +265,8 @@ def handle_mscolab_backend_yaml_init(): want_response_signed: true want_assertion_signed: true allow_unknown_attributes: true - allow_unsolicited: true + # only responses to a login request of this server, see mslib.mscolab.auth.OutstandingSamlRequests + allow_unsolicited: false endpoints: assertion_consumer_service: - [http://localhost:8083/localhost_test_idp/acs/post, 'urn:oasis:names:tc:SAML:2.0:bindings:HTTP-POST'] @@ -321,7 +322,7 @@ def handle_mscolab_backend_yaml_init(): # want_response_signed: true # want_assertion_signed: true # allow_unknown_attributes: true - # allow_unsolicited: true + # allow_unsolicited: false # endpoints: # assertion_consumer_service: # - [http://localhost:8083/idp2/acs/post, 'urn:oasis:names:tc:SAML:2.0:bindings:HTTP-POST'] diff --git a/mslib/msidp/CLAUDE.md b/mslib/msidp/CLAUDE.md index ed021bfeb..014015500 100644 --- a/mslib/msidp/CLAUDE.md +++ b/mslib/msidp/CLAUDE.md @@ -16,8 +16,14 @@ Global map: ../../ARCHITECTURE.md other mslib packages. - Changes here usually pair with `mslib/mscolab/blueprints/auth` (SAML routes) and docs/sso_via_saml_mscolab.rst. +- msidp is test-only and says so on start; it still must not log passwords or + session ids, take a session id from the URL, or render request data unescaped. ## Verify -No dedicated test dir; exercised manually per docs/sso_via_saml_mscolab.rst. -Run `pixi run -e dev lint` and `pixi run -e dev lint-imports`. +`tests/_test_msidp` (sessions, cookies, logs, error messages; WSGI app with a +stand-in pysaml2 server). The full SSO flow with MSColab is exercised manually +per docs/conf_sso_test_msscolab.rst; MSColab's side of it is covered by +`tests/_test_mscolab/test_saml.py` with an in-process pysaml2 IdP (needs +`xmlsec1`, skipped without it). Run `pixi run -e dev lint` and +`pixi run -e dev lint-imports`. diff --git a/mslib/msidp/htdocs/login.mako b/mslib/msidp/htdocs/login.mako index 6d72acd80..e64d78c09 100644 --- a/mslib/msidp/htdocs/login.mako +++ b/mslib/msidp/htdocs/login.mako @@ -5,16 +5,16 @@ To register it's quite simple: enter a valid username and a password

-
- - - + + + +
-
+
@@ -22,7 +22,7 @@
+ value="${password|h}"/>
diff --git a/mslib/msidp/idp.py b/mslib/msidp/idp.py index 81896bcac..03d268218 100644 --- a/mslib/msidp/idp.py +++ b/mslib/msidp/idp.py @@ -32,6 +32,9 @@ import argparse import base64 +import binascii +import hmac +import html import ssl import importlib import logging @@ -44,7 +47,7 @@ from mslib import msidp from http.cookies import SimpleCookie from hashlib import sha1 -from urllib.parse import parse_qs +from urllib.parse import parse_qs, urlencode import saml2.xmldsig as ds from saml2 import ( @@ -56,7 +59,6 @@ server, time_util ) -from saml2.authn import is_equal from saml2.authn_context import ( PASSWORD, UNSPECIFIED, @@ -97,10 +99,44 @@ ) +# how long a login at the IdP is valid, in minutes, in the cookie and on the server +SESSION_LIFETIME = 5 + + class Cache: def __init__(self): self.user2uid = {} self.uid2user = {} + # uid -> time.time() when it expires; the cookie expires too, but its owner could keep sending it + self.uid_expires = {} + + def login(self, user): + """A new session id for user, the one of an earlier login of user ends""" + self.logout(user) + uid = rndstr(24) + self.uid2user[uid] = user + self.user2uid[user] = uid + self.uid_expires[uid] = time.time() + SESSION_LIFETIME * 60 + return uid + + def logout(self, user): + uid = self.user2uid.pop(user, None) + if uid is not None: + self.uid2user.pop(uid, None) + self.uid_expires.pop(uid, None) + + def user_of(self, uid): + """The user of the session uid, None if it is unknown or expired""" + user = self.uid2user.get(uid) + if user is not None and self.uid_expires.get(uid, 0) < time.time(): + self.logout(user) + return None + return user + + +def _error_message(message): + # the text of an error response is HTML; exception texts can contain parts of the request + return html.escape(str(message)) def _expiration(timeout, tformat="%a, %d-%b-%Y %H:%M:%S GMT"): @@ -131,7 +167,6 @@ def dict2list_of_tuples(d): class Service: def __init__(self, environ, start_response, user=None): self.environ = environ - logger.debug("ENVIRON: %s", environ) self.start_response = start_response self.user = user @@ -145,7 +180,6 @@ def unpack_redirect(self): def unpack_post(self): post_data = get_post(self.environ) _dict = parse_qs(post_data if isinstance(post_data, str) else post_data.decode("utf-8")) - logger.debug("unpack_post:: %s", _dict) try: return {k: v[0] for k, v in _dict.items()} except Exception: @@ -357,11 +391,11 @@ def do(self, query, binding_in, relay_state="", encrypt_cert=None, **kwargs): resp_args, _resp = self.verify_request(query, binding_in) except UnknownPrincipal as excp: logger.error("UnknownPrincipal: %s", excp) - resp = ServiceError(f"UnknownPrincipal: {excp}") + resp = ServiceError(_error_message(f"UnknownPrincipal: {excp}")) return resp(self.environ, self.start_response) except UnsupportedBinding as excp: logger.error("UnsupportedBinding: %s", excp) - resp = ServiceError(f"UnsupportedBinding: {excp}") + resp = ServiceError(_error_message(f"UnsupportedBinding: {excp}")) return resp(self.environ, self.start_response) if not _resp: @@ -384,7 +418,7 @@ def do(self, query, binding_in, relay_state="", encrypt_cert=None, **kwargs): ) except Exception as excp: logging.error(exception_trace(excp)) - resp = ServiceError(f"Exception: {excp}") + resp = ServiceError(_error_message(f"Exception: {excp}")) return resp(self.environ, self.start_response) logger.info("AuthNResponse: %s", _resp) @@ -506,17 +540,14 @@ def ecp(self): authz_info = self.environ["HTTP_AUTHORIZATION"] if authz_info.startswith("Basic "): try: - _info = base64.b64decode(authz_info[6:]) - except TypeError: + user, passwd = base64.b64decode(authz_info[6:], validate=True).decode("utf-8").split(":", 1) + except (binascii.Error, UnicodeDecodeError, ValueError): resp = Unauthorized() else: - try: - (user, passwd) = _info.split(":") - if is_equal(PASSWD[user], passwd): - resp = Unauthorized() + if password_is_valid(user, passwd): self.user = user self.environ["idp.authn"] = IdpServerSettings_.AUTHN_BROKER.get_authn_by_accr(PASSWORD) - except ValueError: + else: resp = Unauthorized() else: resp = Unauthorized() @@ -576,15 +607,19 @@ def username_password_authn(environ, start_response, reference, key, redirect_ur "authn_reference": reference, "redirect_uri": redirect_uri, } - logger.info("do_authentication argv: %s", argv) return resp(environ, start_response, **argv) +def password_is_valid(username, password): + """True if password is the one of username in PASSWD""" + return username in PASSWD and hmac.compare_digest(PASSWD[username].encode("utf-8"), password.encode("utf-8")) + + def verify_username_and_password(dic): # verify username and password username = dic["login"][0] password = dic["password"][0] - if PASSWD[username] == password: + if password_is_valid(username, password): return True, username else: return False, None @@ -596,8 +631,6 @@ def do_verify(environ, start_response, _): query_str = query_str.decode("ascii") query = parse_qs(query_str) - logger.debug("do_verify: %s", query) - try: _ok, user = verify_username_and_password(query) except KeyError: @@ -607,15 +640,13 @@ def do_verify(environ, start_response, _): if not _ok: resp = Unauthorized("Unknown user or wrong password") else: - uid = rndstr(24) - IdpServerSettings_.IDP.cache.uid2user[uid] = user - IdpServerSettings_.IDP.cache.user2uid[user] = uid - logger.debug("Register %s under '%s'", user, uid) + uid = IdpServerSettings_.IDP.cache.login(user) + logger.debug("Logged in %s", user) kaka = set_cookie("idpauthn", "/", uid, query["authn_reference"][0]) - lox = f"{query['redirect_uri'][0]}?id={uid}&key={query['key'][0]}" - logger.debug("Redirect => %s", lox) + # the session id only in the cookie: the redirect_uri comes from the form + lox = f"{query['redirect_uri'][0]}?{urlencode({'key': query['key'][0]})}" resp = Redirect(lox, headers=[kaka], content="text/html") return resp(environ, start_response) @@ -649,24 +680,20 @@ def do(self, request, binding, relay_state="", encrypt_cert=None, **kwargs): req_info = IdpServerSettings_.IDP.parse_logout_request(request, binding) except Exception as exc: logger.error("Bad request: %s", exc) - resp = BadRequest(f"{exc}") + resp = BadRequest(_error_message(exc)) return resp(self.environ, self.start_response) msg = req_info.message if msg.name_id: lid = IdpServerSettings_.IDP.ident.find_local_id(msg.name_id) logger.info("local identifier: %s", lid) - if lid in IdpServerSettings_.IDP.cache.user2uid: - uid = IdpServerSettings_.IDP.cache.user2uid[lid] - if uid in IdpServerSettings_.IDP.cache.uid2user: - del IdpServerSettings_.IDP.cache.uid2user[uid] - del IdpServerSettings_.IDP.cache.user2uid[lid] + IdpServerSettings_.IDP.cache.logout(lid) # remove the authentication try: IdpServerSettings_.IDP.session_db.remove_authn_statements(msg.name_id) except KeyError as exc: logger.error("Unknown session: %s", exc) - resp = ServiceError(f"Unknown session: {exc}") + resp = ServiceError(_error_message(f"Unknown session: {exc}")) return resp(self.environ, self.start_response) resp = IdpServerSettings_.IDP.create_logout_response(msg, [binding]) @@ -684,7 +711,7 @@ def do(self, request, binding, relay_state="", encrypt_cert=None, **kwargs): destination, relay_state, response=response) except Exception as exc: logger.error("ServiceError: %s", exc) - resp = ServiceError(f"{exc}") + resp = ServiceError(_error_message(exc)) return resp(self.environ, self.start_response) # _tlh = dict2list_of_tuples(hinfo["headers"]) @@ -746,7 +773,7 @@ def do(self, aid, binding, relay_state="", encrypt_cert=None): try: assertion = IdpServerSettings_.IDP.create_assertion_id_request_response(aid) except Unknown: - resp = NotFound(aid) + resp = NotFound(_error_message(aid)) return resp(self.environ, self.start_response) hinfo = IdpServerSettings_.IDP.apply_binding(BINDING_URI, f"{assertion}", response=True) @@ -871,7 +898,6 @@ def do(self, query, binding, relay_state="", encrypt_cert=None): def info_from_cookie(kaka): - logger.debug("KAKA: %s", kaka) if kaka: cookie_obj = SimpleCookie(kaka) morsel = cookie_obj.get("idpauthn", None) @@ -881,8 +907,11 @@ def info_from_cookie(kaka): if not isinstance(data, str): data = data.decode("ascii") key, ref = data.split(":", 1) - return IdpServerSettings_.IDP.cache.uid2user[key], ref - except (KeyError, TypeError): + user = IdpServerSettings_.IDP.cache.user_of(key) + if user is None: + return None, None + return user, ref + except (KeyError, TypeError, ValueError, binascii.Error, UnicodeDecodeError): return None, None else: logger.debug("No idpauthn cookie") @@ -891,19 +920,24 @@ def info_from_cookie(kaka): def delete_cookie(environ, name): kaka = environ.get("HTTP_COOKIE", "") - logger.debug("delete KAKA: %s", kaka) if kaka: - cookie_obj = SimpleCookie(kaka) - morsel = cookie_obj.get(name, None) cookie = SimpleCookie() cookie[name] = "" - cookie[name]["path"] = "/" - logger.debug("Expire: %s", morsel) + _protect_cookie(cookie[name]) cookie[name]["expires"] = _expiration("dawn") return tuple(cookie.output().split(": ", 1)) return None +def _protect_cookie(morsel): + # not readable by scripts, not sent along with requests of other sites, only by HTTPS if the IdP uses it + morsel["path"] = "/" + morsel["httponly"] = True + morsel["samesite"] = "Lax" + if IdpServerSettings_.https: + morsel["secure"] = True + + def set_cookie(name, _, *args): cookie = SimpleCookie() @@ -916,9 +950,8 @@ def set_cookie(name, _, *args): data64 = data64.decode("ascii") cookie[name] = data64 - cookie[name]["path"] = "/" - cookie[name]["expires"] = _expiration(5) # 5 minutes from now - logger.debug("Cookie expires: %s", cookie[name]["expires"]) + _protect_cookie(cookie[name]) + cookie[name]["expires"] = _expiration(SESSION_LIFETIME) return tuple(cookie.output().split(": ", 1)) @@ -1035,18 +1068,12 @@ def application(environ, start_response): kaka = environ.get("HTTP_COOKIE", None) logger.info(" PATH: %s", path) + # the session only from the cookie, a session id in the URL ends up in logs and the Referer header + user = None if kaka: - logger.info("= KAKA =") user, authn_ref = info_from_cookie(kaka) if authn_ref: environ["idp.authn"] = IdpServerSettings_.AUTHN_BROKER[authn_ref] - else: - try: - query = parse_qs(environ["QUERY_STRING"]) - logger.debug("QUERY: %s", query) - user = IdpServerSettings_.IDP.cache.uid2user[query["id"][0]] - except KeyError: - user = None url_patterns = AUTHN_URLS if not user: @@ -1082,16 +1109,22 @@ def __init__(self): self.AUTHN_BROKER = AuthnBroker() self.IDP = None self.args = None + self.https = False IdpServerSettings_ = IdpServerSettings() +TEST_ONLY_WARNING = """ +WARNING: msidp is an identity provider for development and tests of the SAML login of MSColab only. +Never run it where others can reach it, and never configure an MSColab server for others to trust it: +its users and passwords are the ones in mslib/msidp/idp_user.py, known to everyone, and it lacks protections +of a real identity provider (e.g. it accepts unsigned requests and redirects to any redirect_uri). +""" + + def main(): - warnings.warn( - '\033[91mWARNING: msidp is solely for development and ' - 'testing purposes; do not use in production environments.\033[0m' - ) + warnings.warn(f"\033[91m{TEST_ONLY_WARNING}\033[0m") if os.name == "nt": logger.error("The msidp application is not usable on Windows.") return 1 @@ -1143,6 +1176,7 @@ def main(): ssl_context = None _https = "" + IdpServerSettings_.https = bool(CONFIG.HTTPS) if CONFIG.HTTPS: _https = "using HTTPS" # Creating an SSL context diff --git a/mslib/msidp/idp_conf.py b/mslib/msidp/idp_conf.py index a545713db..fedec2ebc 100644 --- a/mslib/msidp/idp_conf.py +++ b/mslib/msidp/idp_conf.py @@ -179,13 +179,14 @@ def sso_dir_path(local_file): "formatter": "simple", }, }, + # DEBUG logs the passwords of logins and the session ids "loggers": { "saml2": { - "level": "DEBUG" + "level": "WARNING" }, }, "root": { - "level": "DEBUG", + "level": "WARNING", "handlers": [ "stderr", ], diff --git a/mslib/msui/mscolab_connect_dialog.py b/mslib/msui/mscolab_connect_dialog.py index 6a83ad3b9..6b0e90997 100644 --- a/mslib/msui/mscolab_connect_dialog.py +++ b/mslib/msui/mscolab_connect_dialog.py @@ -29,6 +29,7 @@ import webbrowser from urllib.parse import urljoin +from PyQt5 import QtCore from PyQt5.QtWidgets import QDialog, QMessageBox from mslib.msui.qt5 import ui_mscolab_connect_dialog as ui_conn @@ -37,9 +38,12 @@ from keyring.errors import NoKeyringError, PasswordSetError, InitError from mslib.mscolab.api import endpoints from mslib.mscolab.api.schemas import ( + IdpLinkRequest, + IdpLinkResponse, IdpLoginAuthRequest, IdpLoginAuthResponse, LoginRequest, + LoginResponse, RegisterRequest, RegisterResponse, StatusResponse, @@ -64,6 +68,8 @@ def __init__(self, parent=None, mscolab=None): # initialize server url as none self.mscolab_server_url = None self.auth = None + # the code of a SAML login whose identity is linked to the account after the next password login + self.idp_link_code = None self.setFixedSize(self.size()) self.stackedWidget.setCurrentWidget(self.httpAuthPage) @@ -116,7 +122,12 @@ def page_switched(self, index): self.newPasswordLe.setText("") self.newConfirmPasswordLe.setText("") - def set_status(self, _type="Error", msg=""): + def set_status(self, _type="Error", msg="", plain_text=False): + """ + Shows msg in the status line; plain_text for a text of the server, which must not be rendered as rich text + with links or images + """ + self.statusLabel.setTextFormat(QtCore.Qt.PlainText if plain_text else QtCore.Qt.AutoText) if _type == "Error": _msg = f"âš  {msg}" self.statusLabel.setOpenExternalLinks(True) @@ -239,6 +250,7 @@ def disconnect_handler(self): self.mscolab_server_url = None self.auth = None + self.idp_link_code = None self.connectBtn.show() self.connectBtn.setFocus() @@ -285,48 +297,112 @@ def login_handler(self): self.loginBtn.setEnabled(True) else: self.save_user_credentials_to_config_file(req.email, req.password) + if self.idp_link_code is not None: + try: + token = LoginResponse.from_text(response.text).token + except (KeyError, TypeError, ValueError) as ex: + # after_login tells the user that the reply can't be read + logging.error("can't read the login response of %s: %s %s", url, type(ex), ex) + else: + self.link_idp_identity(token) self.mscolab.after_login(req.email, self.mscolab_server_url, response) def idp_login_handler(self): """Handle IDP login Button""" + self.idp_link_code = None url_idp_login = urljoin(self.mscolab_server_url, "available_idps") webbrowser.open(url_idp_login, new=2) self.stackedWidget.setCurrentWidget(self.idpAuthPage) def idp_auth_token_submit_handler(self): - """Handle IDP authentication token submission""" + """Exchanges the code that the browser shows after a SAML login for a login token""" url_idp_login_auth = urljoin(self.mscolab_server_url, endpoints.IDP_LOGIN_AUTH) - user_token = self.idpAuthPasswordLe.text() - + code = self.idpAuthPasswordLe.text().strip() try: - idp_req = IdpLoginAuthRequest(token=user_token) - response = requests.post(url_idp_login_auth, json=idp_req.to_json_data(), - timeout=tuple(config_loader(dataset="MSCOLAB_timeout"))) - if response.status_code == 401: - self.set_status("Error", 'Invalid token or token expired. Please try again') - self.stackedWidget.setCurrentWidget(self.loginPage) - - elif response.status_code == 200: - idp_response = IdpLoginAuthResponse.from_text(response.text) - - req = LoginRequest(email=idp_response.user.emailid, password=idp_response.token) - - session = requests.Session() - session.auth = self.auth - session.headers.update({'x-test': 'true'}) - url = urljoin(self.mscolab_server_url, endpoints.TOKEN) + response = requests.post(url_idp_login_auth, json=IdpLoginAuthRequest(token=code).to_json_data(), + auth=self.auth, timeout=tuple(config_loader(dataset="MSCOLAB_timeout"))) + except requests.exceptions.RequestException as error: + logging.error("unexpected error: %s %s %s", type(error), url_idp_login_auth, error) + self.set_status("Error", f'Failed to establish a new connection to "{self.mscolab_server_url}". ' + 'Try again in a moment.') + return + try: + idp_response = IdpLoginAuthResponse.from_text(response.text) \ + if response.status_code in (200, 403, 409) else None + except (KeyError, TypeError, ValueError) as ex: + # e.g. the error page of a proxy + logging.error("can't read the response of %s: %s %s", url_idp_login_auth, type(ex), ex) + self.set_status("Error", "Can't read the response of the MSColab server. It may be incompatible with this " + "version of MSUI.") + return + if response.status_code == 200: + if idp_response.legacy_password: + # a server before 12.0.0: the token is a one-time password for the login at endpoints.TOKEN + self.legacy_idp_login(idp_response) + return + self.mscolab.after_login(idp_response.user.emailid, self.mscolab_server_url, response) + elif response.status_code == 403: + # refused with a reason, e.g. an account whose email isn't confirmed; the code stays valid. Text of the + # server: plain text + self.set_status("Error", idp_response.message or 'Your identity provider login was refused by the server.', + plain_text=True) + elif response.status_code == 409: + # an account with the email of the SAML login exists: its owner links the SAML login to it by logging in + self.stackedWidget.setCurrentWidget(self.loginPage) + self.loginEmailLe.setText(idp_response.emailid) + self.idp_link_code = code + message = "An account with your email exists. Log in with its password once to link your identity " \ + "provider login to it." + if getattr(self.mscolab.server_status, "mail_enabled", None): + # accounts that logged in by SAML before 12.0.0 got a password nobody knows + url_recover_password = urljoin(self.mscolab_server_url, "reset_request") + message += f' If you do not know it, reset it first.' + self.set_status("Info", message) + else: + self.set_status("Error", 'Invalid or expired code. Please log in by your identity provider again.') + self.stackedWidget.setCurrentWidget(self.loginPage) - response = session.post( - url, data=req.to_form_data(), timeout=tuple(config_loader(dataset="MSCOLAB_timeout"))) - response.raise_for_status() - if response.text == "False": - # show status indicating about wrong credentials - self.set_status("Error", 'Invalid token. Please enter correct token') - elif self.mscolab.after_login(req.email, self.mscolab_server_url, response): - self.set_status("Success", 'Succesfully logged into mscolab server') + def legacy_idp_login(self, idp_response): + """The SAML login at an MSColab server before 12.0.0: logs in with the one-time password it handed out""" + req = LoginRequest(email=idp_response.user.emailid, password=idp_response.token) + session = requests.Session() + session.auth = self.auth + url = urljoin(self.mscolab_server_url, endpoints.TOKEN) + try: + response = session.post(url, data=req.to_form_data(), + timeout=tuple(config_loader(dataset="MSCOLAB_timeout"))) + response.raise_for_status() + except requests.exceptions.RequestException as ex: + logging.error("unexpected error: %s %s %s", type(ex), url, ex) + self.set_status("Error", f'Failed to establish a new connection to "{self.mscolab_server_url}". ' + 'Try again in a moment.') + return + if response.text == "False": + self.set_status("Error", 'Invalid or expired code. Please log in by your identity provider again.') + self.stackedWidget.setCurrentWidget(self.loginPage) + return + self.mscolab.after_login(req.email, self.mscolab_server_url, response) - except requests.exceptions.RequestException as error: - logging.error("unexpected error: %s %s %s", type(error), url, error) + def link_idp_identity(self, token): + """Links the SAML login of idp_link_code to the account of token, which just logged in""" + code, self.idp_link_code = self.idp_link_code, None + url = urljoin(self.mscolab_server_url, endpoints.IDP_LINK) + try: + response = requests.post(url, data=IdpLinkRequest(token=token, code=code).to_form_data(), + auth=self.auth, timeout=tuple(config_loader(dataset="MSCOLAB_timeout"))) + response.raise_for_status() + linked = IdpLinkResponse.from_text(response.text).success + except (requests.exceptions.RequestException, ValueError) as ex: + logging.error("unexpected error: %s %s %s", type(ex), url, ex) + linked = False + if linked: + QMessageBox.information(self, "Identity provider login", + "Your identity provider login is linked to your account. Next time you can log " + "in by your identity provider.") + else: + QMessageBox.warning(self, "Identity provider login", + "Your identity provider login couldn't be linked to your account, maybe its code " + "expired. Log in by your identity provider again to link it.") def save_user_credentials_to_config_file(self, emailid, password): try: diff --git a/tests/_test_api/test_schemas.py b/tests/_test_api/test_schemas.py index 64d6d44bb..9ee751bcd 100644 --- a/tests/_test_api/test_schemas.py +++ b/tests/_test_api/test_schemas.py @@ -66,6 +66,8 @@ GetOperationUsersRequest, GetOperationUsersResponse, GetUserResponse, + IdpLinkRequest, + IdpLinkResponse, IdpLoginAuthRequest, IdpLoginAuthResponse, IdpUserInfo, @@ -687,9 +689,20 @@ def test_round_trip(self): class Test_IdpUserInfo: def test_round_trip(self): - user = IdpUserInfo(username="berta", id=1, emailid="berta@example.com") + user = IdpUserInfo(username="berta", id=1, emailid="berta@example.com", fullname="Berta B") assert IdpUserInfo.from_dict(user.to_dict()) == user + def test_older_server_without_fullname(self): + assert IdpUserInfo.from_dict({"username": "berta", "id": 1, "emailid": "berta@example.com"}).fullname == "" + + def test_is_also_a_login_response(self): + # msui's after_login reads the response of IDP_LOGIN_AUTH as a LoginResponse + response = IdpLoginAuthResponse( + success=True, token="abc", user=IdpUserInfo(username="berta", id=1, emailid="b@example.com", fullname="B")) + login = LoginResponse.from_text(response.to_text()) + assert login.token == "abc" + assert login.user == UserInfo(id=1, username="berta", fullname="B") + class Test_IdpLoginAuthResponse: def test_round_trip_success(self): @@ -704,6 +717,46 @@ def test_round_trip_failure(self): def test_to_dict_omits_token_and_user(self): assert IdpLoginAuthResponse(success=False).to_dict() == {"success": False} + def test_round_trip_link_required(self): + response = IdpLoginAuthResponse(success=False, link_required=True, emailid="berta@example.com") + assert response.to_dict() == {"success": False, "link_required": True, "emailid": "berta@example.com"} + assert IdpLoginAuthResponse.from_text(json.dumps(response.to_dict())) == response + + def test_round_trip_refused_with_message(self): + response = IdpLoginAuthResponse(success=False, message="Confirm your email address") + assert response.to_dict() == {"success": False, "message": "Confirm your email address"} + assert IdpLoginAuthResponse.from_text(json.dumps(response.to_dict())) == response + + def test_reply_of_a_server_before_12_is_a_legacy_password(self): + # its token is a one-time password for endpoints.TOKEN, its user has no fullname + old = IdpLoginAuthResponse.from_text(json.dumps( + {"success": True, "token": "password", "user": {"username": "berta", "id": 1, "emailid": "b@x.org"}})) + assert old.legacy_password is True and old.token == "password" + new = IdpLoginAuthResponse(success=True, token="abc", user=IdpUserInfo(username="berta", id=1, + emailid="b@x.org")) + assert IdpLoginAuthResponse.from_text(new.to_text()).legacy_password is False + # the mark decides, not a missing field: a proxy or a later server may send other user fields + unmarked = json.loads(new.to_text()) + del unmarked["login_token"] + assert IdpLoginAuthResponse.from_text(json.dumps(unmarked)).legacy_password is True + + +class Test_IdpLinkRequest: + def test_round_trip(self): + req = IdpLinkRequest(token="token", code="code") + assert req.to_form_data() == {"token": "token", "code": "code"} + assert IdpLinkRequest.from_form(req.to_form_data()) == req + + +class Test_IdpLinkResponse: + @pytest.mark.parametrize("success", [True, False]) + def test_round_trip(self, success): + response = IdpLinkResponse(success=success) + assert IdpLinkResponse.from_text(json.dumps(response.to_dict())) == response + + def test_invalid_login_token(self): + assert IdpLinkResponse.from_text("False") == IdpLinkResponse(success=False) + class Test_ChatMessageInfo: def test_round_trip_without_replies(self): diff --git a/tests/_test_mscolab/test_migrations.py b/tests/_test_mscolab/test_migrations.py index d8ff35596..72614562e 100644 --- a/tests/_test_mscolab/test_migrations.py +++ b/tests/_test_mscolab/test_migrations.py @@ -117,6 +117,9 @@ def test_upgrade_from(revision, iterations, mscolab_app, tmp_path): # Check that no further migration is required flask_migrate.check(directory=migrations_path) actual_data = {name: db.session.execute(table.select()).all() for name, table in db.metadata.tables.items()} + # Tables added by a later version start empty + for name in set(actual_data) - set(expected_data): + assert actual_data.pop(name) == [], name # Check that all tables have the right number of entries with matching ids copied over assert {k: [e[0] for e in v] for k, v in expected_data.items()} == { k: [e[0] for e in v] for k, v in actual_data.items() @@ -235,6 +238,43 @@ def test_repair_of_database_without_token_nonce(mscolab_app): warning.assert_not_called() +def _tables(): + db.session.remove() + return set(sqlalchemy.inspect(db.engine).get_table_names()) + + +def test_downgrade_removes_saml_tables(mscolab_app): + migrations_path = mslib.mscolab.migrations.__path__[0] + with mscolab_app.app_context(): + assert {"idp_identities", "saml_logins"} <= _tables() + flask_migrate.downgrade(directory=migrations_path, revision="922e4d9c94e2") + assert not {"idp_identities", "saml_logins"} & _tables() + flask_migrate.upgrade(directory=migrations_path) + assert {"idp_identities", "saml_logins"} <= _tables() + + +def test_repair_of_database_without_saml_tables(mscolab_app): + """The server start adds the tables to a database that ran an earlier state of the 12.0.0 migration""" + with mscolab_app.app_context(): + mslib.mscolab.mscolab.handle_db_seed() + with db.engine.begin() as conn: + conn.execute(sqlalchemy.text("DROP TABLE idp_identities")) + conn.execute(sqlalchemy.text("DROP TABLE saml_logins")) + assert not {"idp_identities", "saml_logins"} & _tables() + with mock.patch("mslib.mscolab.app.logging.warning") as warning: + initialise_db() + assert any("tables of SAML logins" in call.args[0] for call in warning.call_args_list) + assert {"idp_identities", "saml_logins"} <= _tables() + migrations_path = mslib.mscolab.migrations.__path__[0] + try: + flask_migrate.check(directory=migrations_path) + except SystemExit as e: + assert e.code == 0 + with mock.patch("mslib.mscolab.app.logging.warning") as warning: + initialise_db() + warning.assert_not_called() + + _DELETED = 999999 # the id of a user and of an operation deleted by an older version diff --git a/tests/_test_mscolab/test_saml.py b/tests/_test_mscolab/test_saml.py new file mode 100644 index 000000000..f58a3c145 --- /dev/null +++ b/tests/_test_mscolab/test_saml.py @@ -0,0 +1,759 @@ +# -*- coding: utf-8 -*- +""" + + tests._test_mscolab.test_saml + ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + + Tests for the SAML login: identities linked to accounts, one-time codes, and which responses are accepted. + + The responses come from an identity provider of pysaml2 in the test process, signed like the ones of a real one. + + This file is part of MSS. + + :copyright: Copyright 2026 Reimar Bauer + :license: APACHE-2.0, see LICENSE for details. + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +""" +import base64 +import datetime +import json +import re +import shutil +from pathlib import Path +from urllib.parse import parse_qs, urlparse + +import mock +import pytest +from saml2 import BINDING_HTTP_POST, BINDING_HTTP_REDIRECT +from saml2.client import Saml2Client +from saml2.config import IdPConfig, SPConfig +from saml2.metadata import create_metadata_string +from saml2.saml import NAMEID_FORMAT_PERSISTENT, NAMEID_FORMAT_TRANSIENT, NameID, AUTHN_PASSWORD +from saml2.server import Server + +from mslib.mscolab import auth +from mslib.mscolab.conf import setup_saml2_backend +from mslib.mscolab.models import db, IdpIdentity, SamlLogin, User +from mslib.mscolab.seed import add_user, get_user +from mslib.mscolab.server import create_server_app +from tests.fixtures import _isolated_mscolab_settings + +XMLSEC1 = shutil.which("xmlsec1") +requires_xmlsec1 = pytest.mark.skipif(XMLSEC1 is None, reason="SAML signatures need xmlsec1") + +IDP_NAME = "test_idp" +IDP_ENTITY_ID = "https://idp.example.org/idp.xml" +SP_ENTITY_ID = "https://mscolab.example.org/sp.xml" +ACS_PATH = f"/{IDP_NAME}/acs/post" +ACS_URL = f"http://localhost{ACS_PATH}" +ALICE = {"eduPersonPrincipalName": ["alice@idp.example.org"], "email": ["alice@example.org"], "givenName": ["Alice"]} +LOCAL_USER = ("alice@example.org", "alice", "alice-password", "Alice Local") + + +def _key_and_cert(directory, name): + from cryptography import x509 + from cryptography.hazmat.primitives import hashes, serialization + from cryptography.hazmat.primitives.asymmetric import rsa + from cryptography.x509.oid import NameOID + key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + subject = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, name)]) + now = datetime.datetime.now(tz=datetime.timezone.utc) + cert = (x509.CertificateBuilder().subject_name(subject).issuer_name(subject).public_key(key.public_key()) + .serial_number(x509.random_serial_number()).not_valid_before(now - datetime.timedelta(days=1)) + .not_valid_after(now + datetime.timedelta(days=1)).sign(key, hashes.SHA256())) + key_file, cert_file = directory / f"{name}.key", directory / f"{name}.crt" + key_file.write_bytes(key.private_bytes(serialization.Encoding.PEM, serialization.PrivateFormat.TraditionalOpenSSL, + serialization.NoEncryption())) + cert_file.write_bytes(cert.public_bytes(serialization.Encoding.PEM)) + return str(key_file), str(cert_file) + + +def _sp_config(key_file, cert_file, idp_metadata=None): + config = { + "entityid": SP_ENTITY_ID, + "key_file": key_file, + "cert_file": cert_file, + "xmlsec_binary": XMLSEC1, + "service": {"sp": { + "endpoints": {"assertion_consumer_service": [(ACS_URL, BINDING_HTTP_POST)]}, + "authn_requests_signed": True, + "want_response_signed": True, + "want_assertion_signed": True, + "allow_unknown_attributes": True, + # as in the configurations before 12.0.0: the server must refuse unsolicited responses anyway + "allow_unsolicited": True, + }}, + } + if idp_metadata is not None: + config["metadata"] = {"inline": [idp_metadata]} + return SPConfig().load(config) + + +def _idp_config(key_file, cert_file, sp_metadata=None): + config = { + "entityid": IDP_ENTITY_ID, + "key_file": key_file, + "cert_file": cert_file, + "xmlsec_binary": XMLSEC1, + "service": {"idp": { + "endpoints": {"single_sign_on_service": [("https://idp.example.org/sso", BINDING_HTTP_REDIRECT)]}, + "policy": {"default": {"attribute_restrictions": None, "lifetime": {"minutes": 15}}}, + "name_id_format": [NAMEID_FORMAT_TRANSIENT, NAMEID_FORMAT_PERSISTENT], + }}, + } + if sp_metadata is not None: + config["metadata"] = {"inline": [sp_metadata]} + return IdPConfig().load(config) + + +def _metadata(config): + return create_metadata_string(None, config, 4, None, None, None, None, None).decode("utf-8") + + +class TestIdp: + """An identity provider that answers the AuthnRequests of the server with signed responses""" + __test__ = False + + def __init__(self, directory): + sp_key, sp_cert = _key_and_cert(directory, "sp") + idp_key, idp_cert = _key_and_cert(directory, "idp") + sp_metadata = _metadata(_sp_config(sp_key, sp_cert)) + idp_metadata = _metadata(_idp_config(idp_key, idp_cert)) + self.server = Server(config=_idp_config(idp_key, idp_cert, sp_metadata)) + self.idp_config = { + 'idp_identity_name': IDP_NAME, + 'idp_data': { + 'idp_name': 'Test Identity Provider', + 'saml2client': Saml2Client(_sp_config(sp_key, sp_cert, idp_metadata)), + 'assertion_consumer_endpoints': [ACS_PATH], + 'user_id_attribute': 'eduPersonPrincipalName', + }, + } + + def response(self, in_response_to, identity=ALICE, name_id=None): + """A signed SAMLResponse to the request in_response_to, as the browser posts it to the server""" + if name_id is None: + name_id = NameID(format=NAMEID_FORMAT_TRANSIENT, text="transient-id") + response = self.server.create_authn_response( + identity, in_response_to, ACS_URL, SP_ENTITY_ID, name_id=name_id, + authn={"class_ref": AUTHN_PASSWORD, "authn_auth": IDP_ENTITY_ID}, + sign_response=True, sign_assertion=True) + return base64.b64encode(str(response).encode("utf-8")).decode("ascii") + + +@pytest.fixture(scope="module") +def test_idp(tmp_path_factory): + return TestIdp(tmp_path_factory.mktemp("saml")) + + +@pytest.fixture +def saml_app(tmp_path, test_idp): + settings = _isolated_mscolab_settings(tmp_path) + settings.USE_SAML2 = True + Path(settings.DATA_DIR).mkdir(parents=True, exist_ok=True) + with mock.patch.object(setup_saml2_backend, "CONFIGURED_IDPS", [test_idp.idp_config]): + app = create_server_app(settings) + with app.app_context(): + yield app + db.session.remove() + db.engine.dispose() + + +def _start_login(client): + """Starts a SAML login like the page of available_idps, returns the id of the AuthnRequest""" + response = client.post("/idp_login/", data={"selectedIdentityProvider": IDP_NAME}) + assert response.status_code == 303 + query = parse_qs(urlparse(response.headers["Location"]).query) + assert "Signature" in query + return SamlLogin.query.filter_by(idp_identity_name=IDP_NAME, code_hash=None).order_by(SamlLogin.id.desc()) \ + .first().request_id + + +def _post_response(client, saml_response): + return client.post(ACS_PATH, data={"SAMLResponse": saml_response}) + + +def _code(response): + assert response.status_code == 200, response.text + return re.search(r'([^<]+)', response.text).group(1) + + +def _saml_login(client, test_idp, **kwargs): + """A complete SAML login in the browser, returns the code that it shows""" + return _code(_post_response(client, test_idp.response(_start_login(client), **kwargs))) + + +def _exchange(client, code): + return client.post("/idp_login_auth/", json={"token": code}) + + +def _authorized(client, token): + return client.get("/test_authorized", query_string={"token": token}).text == "True" + + +def test_saml_routes_need_use_saml2(mscolab_app): + client = mscolab_app.test_client() + assert client.get("/available_idps/").status_code == 404 + assert client.post("/idp_login/", data={"selectedIdentityProvider": IDP_NAME}).status_code == 404 + assert _exchange(client, "code").status_code == 404 + assert client.post("/idp_link", data={"token": "token", "code": "code"}).status_code == 404 + + +@requires_xmlsec1 +def test_server_with_saml_starts_and_offers_the_idps(saml_app): + # the assertion consumer services are added to the app: a blueprint can't change after its registration + assert ACS_PATH in {rule.rule for rule in saml_app.url_map.iter_rules()} + response = saml_app.test_client().get("/available_idps/") + assert response.status_code == 200 + assert f'value="{IDP_NAME}"' in response.text + assert saml_app.test_client().get(f"/metadata/{IDP_NAME}").status_code == 200 + assert saml_app.test_client().get("/metadata/unknown").status_code == 404 + + +@requires_xmlsec1 +def test_new_identity_gets_an_account(saml_app, test_idp): + client = saml_app.test_client() + response = _exchange(client, _saml_login(client, test_idp)) + assert response.status_code == 200 + data = json.loads(response.text) + assert data["success"] is True + assert data["user"]["emailid"] == "alice@example.org" and data["user"]["username"] == "Alice" + assert "fullname" in data["user"] + assert _authorized(client, data["token"]) + user = get_user("alice@example.org") + assert user.authentication_backend == IDP_NAME + assert [(i.issuer, i.subject) for i in user.idp_identities] == [(IDP_ENTITY_ID, "alice@idp.example.org")] + # the next login is the same account, also with another email and transient NameID + identity = dict(ALICE, email=["alice@new.example.org"]) + data = json.loads(_exchange(client, _saml_login(client, test_idp, identity=identity)).text) + assert data["user"]["id"] == user.id + assert User.query.count() == 1 + + +@requires_xmlsec1 +def test_saml_login_needs_a_persistent_identifier(saml_app, test_idp): + client = saml_app.test_client() + identity = {key: value for key, value in ALICE.items() if key != "eduPersonPrincipalName"} + assert _post_response(client, test_idp.response(_start_login(client), identity=identity)).status_code == 403 + # without user_id_attribute the NameID, if it is persistent + del test_idp.idp_config['idp_data']['user_id_attribute'] + try: + assert _post_response(client, test_idp.response(_start_login(client), identity=identity)).status_code == 403 + name_id = NameID(format=NAMEID_FORMAT_PERSISTENT, text="persistent-alice") + code = _saml_login(client, test_idp, identity=identity, name_id=name_id) + finally: + test_idp.idp_config['idp_data']['user_id_attribute'] = 'eduPersonPrincipalName' + assert _exchange(client, code).status_code == 200 + assert [i.subject for i in get_user("alice@example.org").idp_identities] == ["persistent-alice"] + + +@requires_xmlsec1 +def test_saml_login_does_not_take_over_an_account_by_email(saml_app, test_idp): + assert add_user(*LOCAL_USER) + client = saml_app.test_client() + password_hash = get_user(LOCAL_USER[0]).password + code = _saml_login(client, test_idp) + response = _exchange(client, code) + assert response.status_code == 409 + assert json.loads(response.text) == {"success": False, "link_required": True, "emailid": LOCAL_USER[0]} + user = get_user(LOCAL_USER[0]) + assert user.idp_identities == [] and user.authentication_backend == "local" and user.password == password_hash + + # the owner logs in with the password and links the identity + token = json.loads(client.post("/token", data={"email": LOCAL_USER[0], "password": LOCAL_USER[2]}).text)["token"] + assert client.post("/idp_link", data={"token": "invalid", "code": code}).text == "False" + assert client.post("/idp_link", data={"token": token, "code": "invalid"}).get_json() == {"success": False} + assert client.post("/idp_link", data={"token": token, "code": code}).get_json() == {"success": True} + # the code worked once + assert client.post("/idp_link", data={"token": token, "code": code}).get_json() == {"success": False} + assert _exchange(client, code).status_code == 401 + + data = json.loads(_exchange(client, _saml_login(client, test_idp)).text) + assert data["user"]["id"] == user.id + # a SAML login never changes the password + assert get_user(LOCAL_USER[0]).password == password_hash + assert client.post("/token", data={"email": LOCAL_USER[0], "password": LOCAL_USER[2]}).text != "False" + + +@requires_xmlsec1 +def test_only_the_account_of_the_email_can_link(saml_app, test_idp): + assert add_user(*LOCAL_USER) + assert add_user("mallory@example.org", "mallory", "mallory-password", "Mallory") + client = saml_app.test_client() + code = _saml_login(client, test_idp) + assert _exchange(client, code).status_code == 409 + token = json.loads(client.post("/token", data={"email": "mallory@example.org", + "password": "mallory-password"}).text)["token"] + assert client.post("/idp_link", data={"token": token, "code": code}).get_json() == {"success": False} + assert IdpIdentity.query.count() == 0 + + +def _saml_login_of(identity, issuer=IDP_ENTITY_ID, idp_identity_name=IDP_NAME, request_id="id-1"): + """The code of a SAML login of identity, without the IdP""" + auth.remember_saml_request(request_id, idp_identity_name) + return auth.finish_saml_login(request_id, idp_identity_name, issuer, identity["eduPersonPrincipalName"][0], + identity["email"][0], identity["givenName"][0]) + + +@pytest.fixture +def request_idp(monkeypatch): + """The entity id MSColab sends the login requests of IDP_NAME to, without a configured IdP""" + monkeypatch.setattr(auth, "get_idp_entity_id", lambda name: IDP_ENTITY_ID if name == IDP_NAME else None) + + +def test_account_of_an_older_version_is_linked_on_its_first_login(mscolab_app, request_idp): + with mscolab_app.app_context(): + # logged in by this IdP before, by email alone + assert add_user(*LOCAL_USER) + user = get_user(LOCAL_USER[0]) + user.authentication_backend = IDP_NAME + db.session.commit() + # another IdP of the same metadata, e.g. of a federation, can assert the email too: the owner links it + code = _saml_login_of(ALICE, issuer="https://evil.example.org/idp.xml", request_id="id-0") + assert auth.idp_login(code) == (None, LOCAL_USER[0]) + assert user.idp_identities == [] + assert auth.idp_login(_saml_login_of(ALICE, request_id="id-1")) == (user, None) + assert [i.subject for i in user.idp_identities] == ["alice@idp.example.org"] + # from then on only this identity: another one with the same email must be linked by the owner + other = dict(ALICE, eduPersonPrincipalName=["someone@idp.example.org"]) + assert auth.idp_login(_saml_login_of(other, request_id="id-2")) == (None, LOCAL_USER[0]) + + +@pytest.mark.parametrize("backend", ["local", "other_idp"]) +def test_account_of_another_backend_is_not_linked_on_login(mscolab_app, backend): + with mscolab_app.app_context(): + assert add_user(*LOCAL_USER) + user = get_user(LOCAL_USER[0]) + user.authentication_backend = backend + db.session.commit() + code = _saml_login_of(ALICE) + assert auth.idp_login(code) == (None, LOCAL_USER[0]) + assert user.idp_identities == [] + # the code stays valid for the link + assert auth.link_idp_identity(user, code) is True + assert auth.idp_login(_saml_login_of(ALICE, request_id="id-2")) == (user, None) + + +def test_identities_of_different_issuers_are_different(mscolab_app): + with mscolab_app.app_context(): + user, _ = auth.idp_login(_saml_login_of(ALICE, request_id="id-1")) + # another IdP of the same configured metadata, e.g. of a federation, asserts the same subject + code = _saml_login_of(ALICE, issuer="https://evil.example.org/idp.xml", request_id="id-2") + assert auth.idp_login(code) == (None, user.emailid) + + +def test_code_works_once_and_expires(mscolab_app): + with mscolab_app.app_context(): + code = _saml_login_of(ALICE, request_id="id-1") + user, _ = auth.idp_login(code) + assert user is not None + assert auth.idp_login(code) == (None, None) + code = _saml_login_of(ALICE, request_id="id-2") + SamlLogin.query.update({SamlLogin.created_on: datetime.datetime.now(tz=datetime.timezone.utc) + - datetime.timedelta(seconds=auth.IDP_LOGIN_CODE_LIFETIME + 1)}) + db.session.commit() + assert auth.idp_login(code) == (None, None) + assert auth.idp_login(None) == (None, None) and auth.idp_login("") == (None, None) + + +def test_saml_request_is_answered_once_and_in_time(mscolab_app): + with mscolab_app.app_context(): + auth.remember_saml_request("id-1", IDP_NAME) + outstanding = auth.OutstandingSamlRequests(IDP_NAME) + assert outstanding and "id-1" in outstanding and "id-2" not in outstanding and None not in outstanding + assert "id-1" not in auth.OutstandingSamlRequests("other_idp") + assert auth.finish_saml_login("id-1", "other_idp", IDP_ENTITY_ID, "subject", None, None) is None + assert auth.finish_saml_login(None, IDP_NAME, IDP_ENTITY_ID, "subject", None, None) is None + assert auth.finish_saml_login("id-1", IDP_NAME, IDP_ENTITY_ID, "subject", None, None) is not None + # one instance checks one response, it looks up each request once + assert "id-1" in outstanding + assert "id-1" not in auth.OutstandingSamlRequests(IDP_NAME) + assert auth.finish_saml_login("id-1", IDP_NAME, IDP_ENTITY_ID, "subject", None, None) is None + auth.remember_saml_request("id-2", IDP_NAME) + SamlLogin.query.filter_by(request_id="id-2").update({SamlLogin.created_on: datetime.datetime.now( + tz=datetime.timezone.utc) - datetime.timedelta(seconds=auth.SAML_REQUEST_LIFETIME + 1)}) + db.session.commit() + assert "id-2" not in auth.OutstandingSamlRequests(IDP_NAME) + assert auth.finish_saml_login("id-2", IDP_NAME, IDP_ENTITY_ID, "subject", None, None) is None + + +def test_old_saml_logins_are_removed(mscolab_app): + with mscolab_app.app_context(): + auth.remember_saml_request("id-1", IDP_NAME) + SamlLogin.query.update({SamlLogin.created_on: datetime.datetime.now(tz=datetime.timezone.utc) + - datetime.timedelta(days=1)}) + db.session.commit() + auth.remember_saml_request("id-2", IDP_NAME) + assert [login.request_id for login in SamlLogin.query.all()] == ["id-2"] + + +def test_saml_login_without_email_gets_no_account(mscolab_app): + with mscolab_app.app_context(): + auth.remember_saml_request("id-1", IDP_NAME) + code = auth.finish_saml_login("id-1", IDP_NAME, IDP_ENTITY_ID, "subject", None, "Alice") + assert auth.idp_login(code) == (None, None) + assert User.query.count() == 0 + + +def test_deleting_a_user_deletes_its_identities(mscolab_app, mscolab_managers): + _, _, fm = mscolab_managers + with mscolab_app.app_context(): + user, _ = auth.idp_login(_saml_login_of(ALICE)) + assert IdpIdentity.query.count() == 1 + assert fm.modify_user(user, action="delete") + assert IdpIdentity.query.count() == 0 + + +@requires_xmlsec1 +def test_response_is_accepted_once(saml_app, test_idp): + client = saml_app.test_client() + saml_response = test_idp.response(_start_login(client)) + assert _post_response(client, saml_response).status_code == 200 + assert _post_response(client, saml_response).status_code == 403 + + +@requires_xmlsec1 +def test_unsolicited_responses_are_refused(saml_app, test_idp): + client = saml_app.test_client() + _start_login(client) + # to a request this server didn't send, and to none, although the SP configuration allows unsolicited ones + assert _post_response(client, test_idp.response("id-unknown")).status_code == 403 + assert _post_response(client, test_idp.response(None)).status_code == 403 + assert SamlLogin.query.filter(SamlLogin.code_hash.isnot(None)).count() == 0 + + +@requires_xmlsec1 +def test_late_response_is_refused(saml_app, test_idp): + client = saml_app.test_client() + request_id = _start_login(client) + SamlLogin.query.update({SamlLogin.created_on: datetime.datetime.now(tz=datetime.timezone.utc) + - datetime.timedelta(seconds=auth.SAML_REQUEST_LIFETIME + 1)}) + db.session.commit() + assert _post_response(client, test_idp.response(request_id)).status_code == 403 + + +@requires_xmlsec1 +def test_changed_response_is_refused(saml_app, test_idp): + client = saml_app.test_client() + saml_response = base64.b64decode(test_idp.response(_start_login(client))).decode("utf-8") + forged = saml_response.replace("alice@idp.example.org", "bob@idp.example.org") + assert forged != saml_response + assert _post_response(client, base64.b64encode(forged.encode("utf-8")).decode("ascii")).status_code == 403 + assert _post_response(client, "not a SAML response").status_code == 403 + assert IdpIdentity.query.count() == 0 + + +def test_identity_is_the_one_of_the_signed_assertion(): + # pysaml2 checks the issuer of the response around the assertion only if the response is signed + assertion = mock.Mock(issuer=mock.Mock(text=IDP_ENTITY_ID)) + authn_response = mock.Mock(assertion=assertion, name_id=None) + authn_response.issuer.return_value = "https://evil.example.org/idp.xml" + idp_data = {"user_id_attribute": "eduPersonPrincipalName"} + attributes = {"eduPersonPrincipalName": ["alice@idp.example.org"]} + assert auth.saml_identity(authn_response, idp_data, attributes) == (IDP_ENTITY_ID, "alice@idp.example.org") + assertion.issuer = None + assert auth.saml_identity(authn_response, idp_data, attributes) is None + + +def _basic_auth(password="testvaluepassword"): + credentials = base64.b64encode(f"user:{password}".encode("utf-8")).decode("ascii") + return {"Authorization": f"Basic {credentials}"} + + +def test_code_exchange_needs_the_basic_auth_of_the_server(mscolab_app): + mscolab_app.config.update(USE_SAML2=True, ENABLE_BASIC_HTTP_AUTHENTICATION=True) + client = mscolab_app.test_client() + with mscolab_app.app_context(): + code = _saml_login_of(ALICE) + assert _exchange(client, code).status_code == 401 + assert client.post("/idp_login_auth/", json={"token": code}, headers=_basic_auth("wrong")).status_code == 401 + response = client.post("/idp_login_auth/", json={"token": code}, headers=_basic_auth()) + assert response.status_code == 200 + token = json.loads(response.text)["token"] + assert client.post("/idp_link", data={"token": token, "code": code}).status_code == 401 + assert client.post("/idp_link", data={"token": token, "code": code}, + headers=_basic_auth()).get_json() == {"success": False} + + +def test_unconfirmed_account_does_not_log_in_by_saml(mscolab_app, request_idp): + mscolab_app.config.update(USE_SAML2=True, MAIL_ENABLED=True) + client = mscolab_app.test_client() + with mscolab_app.app_context(): + # an account of an older version that logged in by this IdP, with email confirmation like /token + assert add_user(*LOCAL_USER) + user = get_user(LOCAL_USER[0]) + user.authentication_backend = IDP_NAME + user.confirmed = False + db.session.commit() + code = _saml_login_of(ALICE) + response = _exchange(client, code) + # refused before the code is used: nothing is linked yet, the user learns why + assert response.status_code == 403 and "token" not in json.loads(response.text) + assert "isn't confirmed" in json.loads(response.text)["message"] + with mscolab_app.app_context(): + assert get_user(LOCAL_USER[0]).idp_identities == [] + user = get_user(LOCAL_USER[0]) + user.confirmed = True + db.session.commit() + # confirmed, the same code logs in + assert _exchange(client, code).status_code == 200 + + +def test_email_of_an_account_matches_in_any_case(mscolab_app): + with mscolab_app.app_context(): + assert add_user("Alice@Example.org", *LOCAL_USER[1:]) + user = get_user("Alice@Example.org") + code = _saml_login_of(ALICE) + assert auth.idp_login(code) == (None, "Alice@Example.org") + assert auth.link_idp_identity(user, code) is True + assert auth.idp_login(_saml_login_of(ALICE, request_id="id-2")) == (user, None) + assert User.query.count() == 1 + + +def test_failed_link_keeps_the_code(mscolab_app): + with mscolab_app.app_context(): + assert add_user(*LOCAL_USER) + user = get_user(LOCAL_USER[0]) + # two logins of the same identity, both refused until it is linked + first, second = _saml_login_of(ALICE, request_id="id-1"), _saml_login_of(ALICE, request_id="id-2") + assert auth.idp_login(first) == (None, LOCAL_USER[0]) and auth.idp_login(second) == (None, LOCAL_USER[0]) + assert auth.link_idp_identity(user, first) is True + # linked meanwhile: the second link fails, its code still logs in + assert auth.link_idp_identity(user, second) is False + assert auth.idp_login(second) == (user, None) + assert auth.idp_login(second) == (None, None) + + +def test_account_of_an_older_version_gets_only_one_identity_on_login(mscolab_app, request_idp): + with mscolab_app.app_context(): + assert add_user(*LOCAL_USER) + user = get_user(LOCAL_USER[0]) + user.authentication_backend = IDP_NAME + db.session.commit() + first = auth._saml_login_of_code(_saml_login_of(ALICE, request_id="id-1")) + other = dict(ALICE, eduPersonPrincipalName=["someone@idp.example.org"]) + second_code = _saml_login_of(other, request_id="id-2") + second = auth._saml_login_of_code(second_code) + # both found the account without an identity, the first one links + assert auth._use_saml_login(first, user, link=True, first_identity=True) is True + assert auth._use_saml_login(second, user, link=True, first_identity=True) is False + assert [i.subject for i in user.idp_identities] == ["alice@idp.example.org"] + # the code of the second login is kept, for the owner to link it + assert auth.link_idp_identity(user, second_code) is True + + +def test_open_saml_logins_are_capped(mscolab_app, monkeypatch): + monkeypatch.setattr(auth, "MAX_SAML_LOGINS", 2) + with mscolab_app.app_context(): + assert auth.remember_saml_request("id-1", IDP_NAME) is True + assert auth.finish_saml_login("id-1", IDP_NAME, IDP_ENTITY_ID, "subject", None, None) is not None + assert auth.remember_saml_request("id-2", IDP_NAME) is True + # a code that wasn't exchanged yet counts too + assert auth.remember_saml_request("id-3", IDP_NAME) is False + assert SamlLogin.query.count() == 2 + # an expired request makes room + SamlLogin.query.filter_by(request_id="id-2").update({SamlLogin.created_on: datetime.datetime.now( + tz=datetime.timezone.utc) - datetime.timedelta(seconds=auth.SAML_REQUEST_LIFETIME + 1)}) + db.session.commit() + assert auth.remember_saml_request("id-3", IDP_NAME) is True + assert sorted(login.request_id for login in SamlLogin.query.all()) == ["id-1", "id-3"] + + +@requires_xmlsec1 +def test_too_many_open_saml_logins_answer_503(saml_app, monkeypatch): + monkeypatch.setattr(auth, "MAX_SAML_LOGINS", 0) + response = saml_app.test_client().post("/idp_login/", data={"selectedIdentityProvider": IDP_NAME}) + assert response.status_code == 503 + assert SamlLogin.query.count() == 0 + + +@pytest.mark.parametrize("emailid, username, message", [ + (None, "Alice", "no valid email"), + ("not an email", "Alice", "no valid email"), + ("alice@" + "a" * 250 + ".org", "Alice", "no valid email"), + # the start of the email is taken too + ("taken@example.org", "Alice@Example", "are taken or can't be used"), + ("taken@example.org", "A" * 256, "are taken or can't be used"), + ("taken@example.org", "taken", "are taken or can't be used"), + ("taken@example.org", None, "are taken or can't be used"), +], ids=["no-email", "invalid-email", "long-email", "username-with-at", "long-username", "taken-username", + "taken-email-start"]) +def test_saml_login_with_unusable_data_is_refused(mscolab_app, emailid, username, message): + with mscolab_app.app_context(): + assert add_user("bob@example.org", "taken", "bob-password", "Bob") + with pytest.raises(auth.SamlLoginRefused, match=message): + auth.check_saml_login(IDP_ENTITY_ID, "alice@idp.example.org", emailid, username) + + +def test_saml_login_data_is_checked_only_for_a_new_account(mscolab_app): + with mscolab_app.app_context(): + assert auth.check_saml_login(IDP_ENTITY_ID, "subject", "alice@example.org", None) == ( + "alice@example.org", "alice") + assert add_user("Alice@example.org", "taken", "alice-password", "Alice") + # the account of the email is linked, not created + assert auth.check_saml_login(IDP_ENTITY_ID, "subject", "alice@example.org", "taken") == ( + "alice@example.org", None) + assert auth.idp_login(_saml_login_of(dict(ALICE, email=["bob@example.org"], givenName=["Bob"])))[0] is not None + # a linked identity logs in whatever the IdP sends + assert auth.check_saml_login(IDP_ENTITY_ID, "alice@idp.example.org", None, "x@y") == (None, None) + + +@pytest.mark.parametrize("username, expected", [ + ("Alice", "Alice"), ("Max", "alice"), ("Alice@Example", "alice"), ("A" * 256, "alice"), (None, "alice"), +], ids=["given-name", "given-name-taken", "given-name-with-at", "given-name-too-long", "no-given-name"]) +def test_new_account_falls_back_to_the_start_of_the_email(mscolab_app, username, expected): + # common first names are taken soon + with mscolab_app.app_context(): + assert add_user("max@example.org", "Max", "max-password", "Max") + assert auth.check_saml_login(IDP_ENTITY_ID, "subject", "alice@example.org", username) == ( + "alice@example.org", expected) + + +def test_usernames_taken_after_the_response_are_refused(mscolab_app): + with mscolab_app.app_context(): + code = _saml_login_of(ALICE) + assert add_user("bob@example.org", "Alice", "bob-password", "Bob") + assert add_user("carol@example.org", "alice", "carol-password", "Carol") + assert auth.idp_login(code) == (None, None) + assert User.query.count() == 2 + + +@requires_xmlsec1 +def test_saml_login_with_taken_usernames_shows_why(saml_app, test_idp): + assert add_user("bob@example.org", "Alice", "bob-password", "Bob") + assert add_user("carol@example.org", "alice", "carol-password", "Carol") + client = saml_app.test_client() + response = _post_response(client, test_idp.response(_start_login(client))) + assert response.status_code == 403 + assert "The usernames 'Alice' and 'alice' that your identity provider sent are taken" \ + in response.text + assert SamlLogin.query.filter(SamlLogin.code_hash.isnot(None)).count() == 0 + + +def _assertion_response(signed_assertion, signed_response, response_issuer): + assertion = mock.Mock(issuer=mock.Mock(text=IDP_ENTITY_ID), signature="signed" if signed_assertion else None) + authn_response = mock.Mock(assertion=assertion, name_id=None, + response=mock.Mock(signature="signed" if signed_response else None)) + authn_response.issuer.return_value = response_issuer + return authn_response + + +@pytest.mark.parametrize("signed_assertion, signed_response, response_issuer, accepted", [ + (True, False, None, True), + (False, True, IDP_ENTITY_ID, True), + # an IdP of the same metadata signs the response around an assertion that claims another IdP + (False, True, "https://evil.example.org/idp.xml", False), + (False, False, IDP_ENTITY_ID, False), +], ids=["signed-assertion", "signed-response-same-issuer", "signed-response-other-issuer", "nothing-signed"]) +def test_issuer_of_an_unsigned_assertion_needs_the_signed_response(signed_assertion, signed_response, response_issuer, + accepted): + idp_data = {"user_id_attribute": "eduPersonPrincipalName"} + attributes = {"eduPersonPrincipalName": ["alice@idp.example.org"]} + identity = auth.saml_identity(_assertion_response(signed_assertion, signed_response, response_issuer), idp_data, + attributes) + assert identity == ((IDP_ENTITY_ID, "alice@idp.example.org") if accepted else None) + + +def test_identity_refusals_are_logged_with_their_reason(): + response = _assertion_response(True, False, None) + with mock.patch("mslib.mscolab.auth.logging.warning") as warning: + assert auth.saml_identity(response, {"user_id_attribute": "id"}, {"id": ["x" * 256]}) is None + assert "longer than 255" in warning.call_args.args[0] + with mock.patch("mslib.mscolab.auth.logging.warning") as warning: + assert auth.saml_identity(response, {"user_id_attribute": "id"}, {}) is None + assert "persistent identifier" in warning.call_args.args[0] + + +def test_email_in_several_cases_is_refused(mscolab_app): + with mscolab_app.app_context(): + assert add_user("Bob@example.org", "bob1", "bob-password", "Bob") + assert add_user("bob@Example.org", "bob2", "bob-password", "Bob") + with pytest.raises(auth.SamlLoginRefused, match="different upper and lower case"): + auth.check_saml_login(IDP_ENTITY_ID, "subject", "BOB@example.org", "Bob") + # an exact match is not ambiguous + assert auth.check_saml_login(IDP_ENTITY_ID, "subject", "Bob@example.org", "Bob") == ("Bob@example.org", None) + + +@pytest.mark.parametrize("body", ['["x"]', '"code"', '42', 'null']) +def test_code_exchange_with_a_body_that_is_no_object(mscolab_app, body): + mscolab_app.config.update(USE_SAML2=True) + response = mscolab_app.test_client().post("/idp_login_auth/", data=body, content_type="application/json") + assert response.status_code == 401 + + +def test_starting_a_saml_login_needs_the_basic_auth_of_the_server(mscolab_app): + # anyone could fill the open SAML logins otherwise + mscolab_app.config.update(USE_SAML2=True, ENABLE_BASIC_HTTP_AUTHENTICATION=True) + client = mscolab_app.test_client() + assert client.get("/available_idps/").status_code == 401 + assert client.post("/idp_login/", data={"selectedIdentityProvider": IDP_NAME}).status_code == 401 + assert client.post("/idp_login/", data={"selectedIdentityProvider": "unknown"}, + headers=_basic_auth()).status_code == 404 + + +def test_password_reset_unlinks_the_saml_identities(mscolab_app, request_idp): + # an IdP that lets its users choose their email: someone created the account with the address of another person, + # who takes it over by a password reset + with mscolab_app.app_context(): + user, _ = auth.idp_login(_saml_login_of(ALICE, request_id="id-1")) + assert len(user.idp_identities) == 1 + assert user.reset_password(user.password, "the-owners-password") + assert get_user(ALICE["email"][0]).idp_identities == [] + assert get_user(ALICE["email"][0]).authentication_backend == "local" + # the identity must be linked again, with the new password, also by the IdP that created the account + assert auth.idp_login(_saml_login_of(ALICE, request_id="id-2")) == (None, ALICE["email"][0]) + + +def _app_with_idp(tmp_path, idp_config): + settings = _isolated_mscolab_settings(tmp_path) + settings.USE_SAML2 = True + Path(settings.DATA_DIR).mkdir(parents=True, exist_ok=True) + with mock.patch.object(setup_saml2_backend, "CONFIGURED_IDPS", [idp_config]): + return create_server_app(settings) + + +def _mock_idp(path, user_id_attribute="eduPersonPrincipalName", name_id_format=NAMEID_FORMAT_TRANSIENT): + client = mock.Mock() + client.config.getattr.return_value = name_id_format + idp_data = {'assertion_consumer_endpoints': [path], 'saml2client': client} + if user_id_attribute: + idp_data['user_id_attribute'] = user_id_attribute + return {'idp_identity_name': 'mocked_idp', 'idp_data': idp_data} + + +@pytest.mark.parametrize("path", ["mocked_idp/acs/post/", "mocked_idp/acs/post"], ids=["trailing-slash", "no-slash"]) +def test_assertion_consumer_service_as_configured(tmp_path, path): + # the metadata tells the IdP the configured URL, with its trailing slash + app = _app_with_idp(tmp_path, _mock_idp(path)) + adapter = app.url_map.bind("localhost") + for url in ("/mocked_idp/acs/post/", "/mocked_idp/acs/post"): + assert adapter.match(url, method="POST")[0] == "saml_acs_mocked_idp_0" + + +def test_idp_without_identifier_warns_at_the_start(tmp_path): + with mock.patch("mslib.mscolab.blueprints.auth.logging.warning") as warning: + _app_with_idp(tmp_path, _mock_idp("mocked_idp/acs/post", user_id_attribute=None)) + assert any("every SAML login by it is refused" in call.args[0] for call in warning.call_args_list) + with mock.patch("mslib.mscolab.blueprints.auth.logging.warning") as warning: + _app_with_idp(tmp_path, _mock_idp("mocked_idp/acs/post", user_id_attribute=None, + name_id_format=[NAMEID_FORMAT_PERSISTENT])) + assert not any("every SAML login by it is refused" in call.args[0] for call in warning.call_args_list) + + +@pytest.mark.parametrize("username", [" ", " Alice "], ids=["blank", "padded"]) +def test_given_name_is_stripped(mscolab_app, username): + with mscolab_app.app_context(): + expected = username.strip() or "alice" + assert auth.check_saml_login(IDP_ENTITY_ID, "subject", "alice@example.org", username) == ( + "alice@example.org", expected) diff --git a/tests/_test_mscolab/test_token_hygiene.py b/tests/_test_mscolab/test_token_hygiene.py index 2b6177bcf..ee90bfb73 100644 --- a/tests/_test_mscolab/test_token_hygiene.py +++ b/tests/_test_mscolab/test_token_hygiene.py @@ -131,16 +131,12 @@ def test_logout_everywhere(app): def test_token_purposes_are_separate(app): email = USER[0] confirmation = auth.generate_confirmation_token(email, auth.EMAIL_CONFIRMATION) - idp_login = auth.generate_confirmation_token(email, auth.IDP_LOGIN) reset = auth.generate_password_reset_token(get_user(email)) assert auth.confirm_token(confirmation, auth.EMAIL_CONFIRMATION) == email - assert auth.confirm_token(idp_login, auth.IDP_LOGIN) == email # a token of one purpose is no token of another one - assert auth.confirm_token(confirmation, auth.IDP_LOGIN) is False - assert auth.confirm_token(idp_login, auth.EMAIL_CONFIRMATION) is False + assert auth.confirm_token(confirmation, auth.PASSWORD_RESET) is False assert auth.confirm_token(reset, auth.EMAIL_CONFIRMATION) is False assert auth.confirm_password_reset_token(confirmation) is None - assert auth.confirm_password_reset_token(idp_login) is None assert auth.confirm_password_reset_token(reset)[0].emailid == email with pytest.raises(ValueError): auth.generate_confirmation_token(email, "unknown") diff --git a/tests/_test_msidp/test_msidp.py b/tests/_test_msidp/test_msidp.py new file mode 100644 index 000000000..1c4894348 --- /dev/null +++ b/tests/_test_msidp/test_msidp.py @@ -0,0 +1,177 @@ +# -*- coding: utf-8 -*- +""" + + tests._test_msidp.test_msidp + ~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + + Tests for the sessions, cookies, logs and error messages of the test identity provider msidp. + + This file is part of MSS. + + :copyright: Copyright 2026 Reimar Bauer + :license: APACHE-2.0, see LICENSE for details. + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +""" +import base64 +import logging +import shutil + +import mock +import pytest +from werkzeug.test import Client + +from mslib.msidp import idp +from mslib.msidp.idp_user import PASSWD + +USER, PASSWORD = next(iter(PASSWD.items())) +ENVIRON = {"REQUEST_METHOD": "POST", "CONTENT_LENGTH": "0", "wsgi.url_scheme": "https", "HTTP_HOST": "localhost", + "PATH_INFO": "/sso/ecp"} + + +@pytest.fixture +def server(): + """msidp with a stand-in for the pysaml2 server, only its session cache is real""" + with mock.patch.object(idp.IdpServerSettings_, "IDP", mock.Mock(cache=idp.Cache())) as idp_server, \ + mock.patch.object(idp.IdpServerSettings_, "AUTHN_BROKER", mock.MagicMock()), \ + mock.patch.object(idp.IdpServerSettings_, "https", False): + yield idp_server + + +@pytest.fixture +def probe(): + """Replaces the SAML endpoints by one that records the user msidp found for a request""" + users = [] + + def callback(environ, start_response, user): + users.append(user) + start_response("200 OK", [("Content-Type", "text/plain")]) + return [b"probe"] + + with mock.patch.object(idp, "AUTHN_URLS", [(r"probe$", callback)]): + yield users + + +def _verify(client, password=PASSWORD): + form = {"login": USER, "password": password, "key": "the-key", "authn_reference": "ref", + "redirect_uri": "https://sp.example.org/sso"} + return client.post("/verify", data=form) + + +def test_login_sets_a_protected_cookie_and_no_session_id_in_the_url(server): + response = _verify(Client(idp.application)) + assert response.status_code == 302 + assert response.headers["Location"] == "https://sp.example.org/sso?key=the-key" + cookie = response.headers["Set-Cookie"] + assert "HttpOnly" in cookie and "SameSite=Lax" in cookie and "Secure" not in cookie + uid = server.cache.user2uid[USER] + assert uid not in response.headers["Location"] + with mock.patch.object(idp.IdpServerSettings_, "https", True): + assert "Secure" in _verify(Client(idp.application)).headers["Set-Cookie"] + + +def test_wrong_password_or_unknown_user_is_refused(server): + assert _verify(Client(idp.application), password="wrong").status_code == 401 + response = Client(idp.application).post("/verify", data={"login": "nobody", "password": "x", "key": "k", + "authn_reference": "r", "redirect_uri": "u"}) + assert response.status_code == 401 + assert server.cache.uid2user == {} + + +def test_session_only_from_the_cookie(server, probe): + client = Client(idp.application) + assert _verify(client).status_code == 302 + uid = server.cache.user2uid[USER] + client.get("/probe") + # the same session id in the URL, without the cookie + Client(idp.application).get("/probe", query_string={"id": uid}) + assert probe == [USER, None] + + +def test_session_expires_on_the_server(server, probe): + client = Client(idp.application) + assert _verify(client).status_code == 302 + expired = idp.time.time() + idp.SESSION_LIFETIME * 60 + 1 + with mock.patch.object(idp.time, "time", return_value=expired): + client.get("/probe") + assert probe == [None] + assert server.cache.uid2user == {} and server.cache.user2uid == {} + + +def test_new_login_ends_the_earlier_session(server): + first = server.cache.login(USER) + second = server.cache.login(USER) + assert server.cache.user_of(first) is None + assert server.cache.user_of(second) == USER + server.cache.logout(USER) + assert server.cache.user_of(second) is None + + +def test_error_messages_are_escaped(server): + server.parse_logout_request.side_effect = ValueError('') + environ = {"REQUEST_METHOD": "GET", "QUERY_STRING": ""} + start_response = mock.Mock() + body = b"".join(idp.SLO(environ, start_response).do("request", idp.BINDING_HTTP_REDIRECT)) + assert start_response.call_args.args[0].startswith("400") + assert b"