Merge branch 'main' into lundberg_resetpw_captcha_and_mfa_security_key

requirements/sub_main.txt

@@ -18,13 +18,13 @@ importlib-metadata==8.5.0 \
     --hash=sha256:45e54197d28b7a7f1559e60b95e7c567032b602131fbd588f1497f47880aa68b \
     --hash=sha256:71522656f0abace1d072b9e5481a48f07c138e00f079c38c8f883823f9c26bd7
     # via -r sub_main.in
-importlib-resources==6.4.5 \
-    --hash=sha256:980862a1d16c9e147a59603677fa2aa5fd82b87f223b6cb870695bcfce830065 \
-    --hash=sha256:ac29d5f956f01d5e4bb63102a5a19957f1b9175e45649977264a1416783bb717
+importlib-resources==6.5.2 \
+    --hash=sha256:185f87adef5bcc288449d98fb4fba07cea78bc036455dd44c5fc4a2fe78fed2c \
+    --hash=sha256:789cfdc3ed28c78b67a06acb8126751ced69a3d5f79c095a98298cd8a760ccec
     # via -r sub_main.in
-setuptools==75.6.0 \
-    --hash=sha256:8199222558df7c86216af4f84c30e9b34a61d8ba19366cc914424cdbd28252f6 \
-    --hash=sha256:ce74b49e8f7110f9bf04883b730f4765b774ef3ef28f722cce7c273d253aaf7d
+setuptools==75.8.0 \
+    --hash=sha256:c5afc8f407c626b8313a86e10311dd3f661c6cd9c09d4bf8c15c0e11f9f2b0e6 \
+    --hash=sha256:e3982f444617239225d675215d51f6ba05f845d4eec313da4418fdbb56fb27e3
     # via -r sub_main.in
 tomli==2.2.1 \
     --hash=sha256:023aa114dd824ade0100497eb2318602af309e5a55595f76b626d6d9f3b7b0a6 \

ruff.toml

@@ -3,9 +3,29 @@ line-length = 120
 target-version = "py311"
 
 [lint]
-select = ["E", "F", "W", "I", "ASYNC", "UP", "FLY", "PERF", "FURB", "ERA", "ANN", "PIE", "PL", "PGH", "FAST"]
+select = [
+  "ANN",
+  "ASYNC",
+  "E",
+  "ERA",
+  "F",
+  "FAST",
+  "FLY",
+  "FURB",
+  "I",
+  "PERF",
+  "PGH",
+  "PIE",
+  "PL",
+  "UP",
+  "W",
+]
 
-ignore = ["E501", "PLR", "PLW"]
+ignore = [
+  "E501",
+  "PLR",
+  "PLW",
+]
 
 [lint.flake8-annotations]
 allow-star-arg-any = true

src/eduid/common/config/base.py

@@ -92,10 +92,10 @@ class RootConfig(BaseModel):
 class LoggingFilters(StrEnum):
     """Identifiers to coherently map elements in LocalContext.filters to filter classes in logging dictConfig."""
 
-    DEBUG_TRUE: str = "require_debug_true"
-    DEBUG_FALSE: str = "require_debug_false"
-    NAMES: str = "app_filter"
-    SESSION_USER: str = "user_filter"
+    DEBUG_TRUE = "require_debug_true"
+    DEBUG_FALSE = "require_debug_false"
+    NAMES = "app_filter"
+    SESSION_USER = "user_filter"
 
 
 class WorkerConfig(RootConfig):

src/eduid/common/fastapi/api_router.py

@@ -16,8 +16,7 @@ class APIRouter(FastAPIRouter):
     def api_route(
         self, path: str, *, include_in_schema: bool = True, **kwargs: Any
     ) -> Callable[[DecoratedCallable], DecoratedCallable]:
-        if path.endswith("/"):
-            path = path[:-1]
+        path = path.removesuffix("/")
 
         add_path = super().api_route(path, include_in_schema=include_in_schema, **kwargs)
 

src/eduid/graphdb/db.py

@@ -41,7 +41,7 @@ def __init__(self, db_uri: str, config: Mapping[str, Any] | None = None) -> None
         self._driver = GraphDatabase.driver(self._db_uri, **_config)
 
     def __repr__(self) -> str:
-        return f'<eduID {self.__class__.__name__}: {getattr(self, "_username", None)}@{getattr(self, "_db_uri", None)}>'
+        return f"<eduID {self.__class__.__name__}: {getattr(self, '_username', None)}@{getattr(self, '_db_uri', None)}>"
 
     def count_nodes(self, label: str | None = None) -> int | None:
         match_statement = "MATCH ()"

src/eduid/graphdb/testing.py

@@ -113,9 +113,9 @@ def purge_db(self) -> None:
             s.run(q)
             # Drop constraints and indices
             for constraint in s.run("CALL db.constraints"):
-                s.run(f'DROP CONSTRAINT {constraint["name"]}')
+                s.run(f"DROP CONSTRAINT {constraint['name']}")
             for index in s.run("CALL db.indexes"):
-                s.run(f'DROP INDEX {index["name"]}')
+                s.run(f"DROP INDEX {index['name']}")
 
     @classmethod
     def get_instance(cls: type[Neo4jTemporaryInstance], max_retry_seconds: int = 60) -> Neo4jTemporaryInstance:

src/eduid/queue/workers/mail.py

@@ -99,8 +99,7 @@ async def sendmail(self, sender: str, recipient: str, message: str, reference: s
         if self.config.environment == EduidEnvironment.dev:
             logger.info("sendmail task:")
             logger.info(
-                f"\nType: email\nReference: {reference}\nSender: {sender}\nRecipient: {recipient}\n"
-                f"Message:\n{message}"
+                f"\nType: email\nReference: {reference}\nSender: {sender}\nRecipient: {recipient}\nMessage:\n{message}"
             )
             return Status(success=True, message="Devel message printed")
 

src/eduid/satosa/scimapi/stepup.py

@@ -684,8 +684,7 @@ def process(
                 _asserted_loa: str | None = data.auth_info.auth_class_ref
                 if _asserted_loa in _loa_settings.requested or _asserted_loa in _loa_settings.extra_accepted:
                     logger.info(
-                        "Rewriting authnContextClassRef in response from "
-                        f"{_asserted_loa} to {_loa_settings.returned}"
+                        f"Rewriting authnContextClassRef in response from {_asserted_loa} to {_loa_settings.returned}"
                     )
                     data.auth_info.auth_class_ref = _loa_settings.returned
                     AuthnContext.save_to_state(context=context, state_key=STATE_KEY_LOA, data=[IDP_SENT_LOA])

src/eduid/scimapi/api_router.py

@@ -16,8 +16,7 @@ class APIRouter(FastAPIRouter):
     def api_route(
         self, path: str, *, include_in_schema: bool = True, **kwargs: Any
     ) -> Callable[[DecoratedCallable], DecoratedCallable]:
-        if path.endswith("/"):
-            path = path[:-1]
+        path = path.removesuffix("/")
 
         add_path = super().api_route(path, include_in_schema=include_in_schema, **kwargs)
 

src/eduid/scimapi/context.py

@@ -120,5 +120,5 @@ def check_version(self, req: ContextRequest, db_obj: ScimApiGroup | ScimApiUser
         if req.headers.get("IF-MATCH") == make_etag(db_obj.version):
             return True
         self.logger.error("Version mismatch")
-        self.logger.debug(f'{req.headers.get("IF-MATCH")} != {make_etag(db_obj.version)}')
+        self.logger.debug(f"{req.headers.get('IF-MATCH')} != {make_etag(db_obj.version)}")
         return False

src/eduid/scimapi/test-scripts/scim-util.py

@@ -195,7 +195,7 @@ def put_group(api: Api, scim_id: str, data: dict[str, Any], token: str | None =
         scim["displayName"] = display_name
     if members:
         new_members = [
-            {"$ref": f'{api}/Users/{member["id"]}', "value": member["id"], "display": member["display_name"]}
+            {"$ref": f"{api}/Users/{member['id']}", "value": member["id"], "display": member["display_name"]}
             for member in members
         ]
         scim["members"] = new_members

src/eduid/scimapi/testing.py

@@ -288,9 +288,9 @@ def _assertName(db_name: ScimApiName, response_name: dict[str, str]) -> None:
         ]
         db_name_dict = asdict(db_name)
         for first, second in name_map:
-            assert db_name_dict.get(first) == response_name.get(
-                second
-            ), f"{first}:{db_name_dict.get(first)} != {second}:{response_name.get(second)}"
+            assert db_name_dict.get(first) == response_name.get(second), (
+                f"{first}:{db_name_dict.get(first)} != {second}:{response_name.get(second)}"
+            )
 
     @staticmethod
     def _assertResponse(response: Response, status_code: int = 200) -> None:
@@ -300,6 +300,6 @@ def _assertResponse(response: Response, status_code: int = 200) -> None:
                 _detail = response.json().get("detail", "No error detail in parsed_response")
         except JSONDecodeError:
             pass
-        assert (
-            response.status_code == status_code
-        ), f"Response status was not {status_code} ({response.status_code}), {_detail}"
+        assert response.status_code == status_code, (
+            f"Response status was not {status_code} ({response.status_code}), {_detail}"
+        )

src/eduid/scimapi/tests/test_scimgroup.py

@@ -428,7 +428,7 @@ def test_update_group_not_found(self) -> None:
             "displayName": "Another display name",
             "members": [],
         }
-        response = self.client.put(url=f'/Groups/{req["id"]}', json=req, headers=self.headers)
+        response = self.client.put(url=f"/Groups/{req['id']}", json=req, headers=self.headers)
         self._assertScimError(response.json(), status=404, detail="Group not found")
 
     def test_version_mismatch(self) -> None:

src/eduid/scimapi/tests/test_scimuser.py

@@ -204,19 +204,19 @@ def _assertUserUpdateSuccess(self, req: Mapping, response: Response, user: ScimA
         self._assertScimResponseProperties(response, resource=user, expected_schemas=expected_schemas)
 
         # Validate user update specifics
-        assert user.external_id == response.json().get(
-            "externalId"
-        ), 'user.externalId != parsed_response.json().get("externalId")'
+        assert user.external_id == response.json().get("externalId"), (
+            'user.externalId != parsed_response.json().get("externalId")'
+        )
         self._assertName(user.name, response.json().get("name"))
         _expected_emails = filter_none(normalised_data([email.to_dict() for email in user.emails]))
         _obtained_emails = filter_none(normalised_data(response.json().get("emails", [])))
         assert _obtained_emails == _expected_emails, 'parsed_response.json().get("email") != user.emails'
         _expected_phones = filter_none(normalised_data([number.to_dict() for number in user.phone_numbers]))
         _obtained_phones = filter_none(normalised_data(response.json().get("phoneNumbers", [])))
         assert _obtained_phones == _expected_phones, 'parsed_response.json().get("phoneNumbers") != user.phone_numbers'
-        assert user.preferred_language == response.json().get(
-            "preferredLanguage"
-        ), 'user.preferred_language != parsed_response.json().get("preferredLanguage")'
+        assert user.preferred_language == response.json().get("preferredLanguage"), (
+            'user.preferred_language != parsed_response.json().get("preferredLanguage")'
+        )
 
         # If the request has NUTID profiles, ensure they are present in the parsed_response
         if SCIMSchema.NUTID_USER_V1.value in req:
@@ -473,7 +473,7 @@ def test_update_user_change_properties(self) -> None:
             },
         }
         self.headers["IF-MATCH"] = create_response.headers["etag"]
-        response = self.client.put(url=f'/Users/{create_response.json()["id"]}', json=req, headers=self.headers)
+        response = self.client.put(url=f"/Users/{create_response.json()['id']}", json=req, headers=self.headers)
         self._assertResponse(response)
 
         assert self.userdb

src/eduid/userdb/admin/__init__.py

@@ -83,8 +83,7 @@ def save_with_backup(self, raw: RawData, dry_run: bool = True) -> str | None:
 
         if not os.path.isdir(self._backupbase):
             sys.stderr.write(
-                f"\n\nBackup basedir {self._backupbase} not found, "
-                "running in a container without the volume mounted?\n"
+                f"\n\nBackup basedir {self._backupbase} not found, running in a container without the volume mounted?\n"
             )
             sys.exit(1)
 
@@ -202,8 +201,7 @@ def _make_backupdir(self, db_coll: str, _id: str) -> str:
 
         if not os.path.isdir(self._backupbase):
             sys.stderr.write(
-                f"\n\nBackup basedir {self._backupbase} not found, running in a container "
-                "without the volume mounted?\n"
+                f"\n\nBackup basedir {self._backupbase} not found, running in a container without the volume mounted?\n"
             )
             sys.exit(1)
 

src/eduid/userdb/user.py

@@ -102,7 +102,7 @@ def check_revoked(cls, values: dict[str, Any]) -> dict[str, Any]:
         # raise exception if the user is revoked
         if values.get("revoked_ts") is not None:
             raise UserIsRevoked(
-                f'User {values.get("user_id")}/{values.get("eppn")} was revoked at {values.get("revoked_ts")}'
+                f"User {values.get('user_id')}/{values.get('eppn')} was revoked at {values.get('revoked_ts')}"
             )
         return values
 

src/eduid/vccs/server/db.py

@@ -16,23 +16,23 @@
 
 @unique
 class Status(StrEnum):
-    ACTIVE: str = "active"
-    DISABLED: str = "disabled"
+    ACTIVE = "active"
+    DISABLED = "disabled"
 
 
 @unique
 class Version(StrEnum):
-    NDNv1: str = "NDNv1"
+    NDNv1 = "NDNv1"
 
 
 @unique
 class KDF(StrEnum):
-    PBKDF2_HMAC_SHA512: str = "PBKDF2-HMAC-SHA512"
+    PBKDF2_HMAC_SHA512 = "PBKDF2-HMAC-SHA512"
 
 
 class CredType(StrEnum):
-    PASSWORD: str = "password"
-    REVOKED: str = "revoked"
+    PASSWORD = "password"
+    REVOKED = "revoked"
 
 
 class CredentialPydanticConfig:

src/eduid/vccs/server/endpoints/misc.py

@@ -9,8 +9,8 @@
 @unique
 class Status(StrEnum):
     # STATUS_x_ is less ambiguous when pattern matching than just 'x'
-    OK: str = "STATUS_OK_"
-    FAIL: str = "STATUS_FAIL_"
+    OK = "STATUS_OK_"
+    FAIL = "STATUS_FAIL_"
 
 
 class StatusResponse(BaseModel):

src/eduid/webapp/common/api/schemas/csrf.py

@@ -44,7 +44,7 @@ class CSRFResponseMixin(Schema):
     def get_csrf_token(self, out_data: dict[str, Any], **kwargs: Any) -> dict[str, Any]:
         # Generate a new csrf token for every response
         out_data["csrf_token"] = session.new_csrf_token()
-        logger.debug(f'Generated new CSRF token in CSRFResponseMixin: {out_data["csrf_token"]}')
+        logger.debug(f"Generated new CSRF token in CSRFResponseMixin: {out_data['csrf_token']}")
         return out_data
 
 

src/eduid/webapp/common/api/testing.py

@@ -508,24 +508,24 @@ def _assure_not_in_dict(d: Mapping[str, Any], unwanted_key: str) -> None:
             _json = response.json
             assert _json
             if type_ is not None:
-                assert type_ == _json["type"], f'Wrong response type. expected: {type_}, actual: {_json["type"]}'
+                assert type_ == _json["type"], f"Wrong response type. expected: {type_}, actual: {_json['type']}"
             assert "payload" in _json, 'JSON body has no "payload" element'
             if message is not None:
                 assert "message" in _json["payload"], 'JSON payload has no "message" element'
                 _message_value = _json["payload"]["message"]
-                assert (
-                    message.value == _message_value
-                ), f"Wrong message returned. expected: {message.value}, actual: {_message_value}"
+                assert message.value == _message_value, (
+                    f"Wrong message returned. expected: {message.value}, actual: {_message_value}"
+                )
             if error is not None:
                 assert _json["error"] is True, "The Flux response was supposed to have error=True"
                 assert "error" in _json["payload"], 'JSON payload has no "error" element'
                 _error = _json["payload"]["error"]
                 assert error == _error, f"Wrong error returned. expected: {error}, actual: {_error}"
             if payload is not None:
                 for k, v in payload.items():
-                    assert (
-                        k in _json["payload"]
-                    ), f"The Flux response payload {_json['payload']} does not contain {repr(k)}"
+                    assert k in _json["payload"], (
+                        f"The Flux response payload {_json['payload']} does not contain {repr(k)}"
+                    )
                     assert v == _json["payload"][k], (
                         f"The Flux response payload item {repr(k)} should be {repr(v)} "
                         f"but is {repr(_json['payload'][k])}"
@@ -536,9 +536,9 @@ def _assure_not_in_dict(d: Mapping[str, Any], unwanted_key: str) -> None:
             if meta is not None:
                 for k, v in meta.items():
                     assert k in _json["meta"], f"The Flux response meta does not contain {repr(k)}"
-                    assert (
-                        v == _json["meta"][k]
-                    ), f"The Flux response meta item {repr(k)} should be {repr(v)} but is {repr(_json['meta'][k])}"
+                    assert v == _json["meta"][k], (
+                        f"The Flux response meta item {repr(k)} should be {repr(v)} but is {repr(_json['meta'][k])}"
+                    )
 
         except (AssertionError, KeyError):
             if response.json:

src/eduid/webapp/common/authn/eduid_saml2.py

@@ -183,8 +183,7 @@ def authenticate(session_info: SessionInfo, strip_suffix: str | None, userdb: Us
     # eduPersonPrincipalName might be scoped and the scope (e.g. "@example.com")
     # might have to be removed before looking for the user in the database.
     if strip_suffix:
-        if saml_user.endswith(strip_suffix):
-            saml_user = saml_user[: -len(strip_suffix)]
+        saml_user = saml_user.removesuffix(strip_suffix)
 
     logger.debug(f"Looking for user with eduPersonPrincipalName == {repr(saml_user)}")
     try: