eduid_user_cleaner skatteverket worker

.vscode/settings.json

@@ -17,11 +17,13 @@
     "python.envFile": "${workspaceFolder}/.vscode/.env",
     "cSpell.words": [
         "amapi",
+        "amdb",
         "authlib",
         "authns",
         "baar",
         "backchannel",
         "backupdir",
+        "bson",
         "checkpw",
         "chpass",
         "csrft",
@@ -32,34 +34,46 @@
         "eppn",
         "fastapi",
         "Freja",
+        "gnap",
         "hashname",
         "hashpw",
         "httpx",
         "hubba",
+        "IDNIN",
         "jsonify",
         "jwks",
         "klass",
+        "ladok",
         "lundberg",
         "masv",
         "mischttp",
         "nameid",
         "navet",
+        "nins",
         "NORDUnet",
         "oidc",
+        "orcid",
+        "prid",
         "proofings",
         "proquint",
         "PWAUTH",
+        "pydantic",
         "pysaml2",
         "reauthn",
         "replicaset",
+        "respx",
         "scim",
         "SCIMAPI",
         "sess",
+        "SIGTERM",
         "skatteverket",
         "Svenska",
         "svipe",
         "svipeid",
         "swamid",
+        "swedenconnect",
+        "TELE",
+        "teleadress",
         "Testsson",
         "timeutil",
         "Unmarshal",
@@ -74,4 +88,4 @@
         "zxcvbn"
     ],
     "python.formatting.provider": "black"
-}
+}

requirements/test_requirements.txt

@@ -2044,4 +2044,4 @@ zxcvbn==4.4.28 \
     --hash=sha256:151bd816817e645e9064c354b13544f85137ea3320ca3be1fb6873ea75ef7dc1
     # via
     #   -c main.txt
-    #   -r main.in
+    #   -r main.in

src/eduid/common/clients/amapi_client/amapi_client.py

@@ -20,15 +20,15 @@
 
 
 class AMAPIClient(GNAPClient):
-    def __init__(self, amapi_url: str, app, auth_data=GNAPClientAuthData, **kwargs):
-        super().__init__(auth_data=auth_data, app=app, **kwargs)
+    def __init__(self, amapi_url: str, auth_data=GNAPClientAuthData, verify_tls: bool = True, **kwargs):
+        super().__init__(auth_data=auth_data, verify=verify_tls, **kwargs)
         self.amapi_url = amapi_url
 
     def _users_base_url(self) -> str:
         return urlappend(self.amapi_url, "users")
 
     def _put(self, base_path: str, user: str, endpoint: str, body: Any) -> httpx.Response:
-        return self.put(urlappend(base_path, f"{user}/{endpoint}"), json=body.json())
+        return self.put(url=urlappend(base_path, f"{user}/{endpoint}"), content=body.json())
 
     def update_user_email(self, user: str, body: UserUpdateEmailRequest) -> UserUpdateResponse:
         ret = self._put(base_path=self._users_base_url(), user=user, endpoint="email", body=body)

src/eduid/common/clients/amapi_client/testing.py

@@ -1,19 +1,34 @@
-from typing import Optional
+from typing import Mapping, Optional
 
 import respx
+from httpx import Response, Request
+from eduid.common.models.amapi_user import UserUpdateNameRequest
 
 from eduid.common.clients.gnap_client.testing import MockedSyncAuthAPIMixin
+from eduid.userdb.userdb import AmDB
 
 
 class MockedAMAPIMixin(MockedSyncAuthAPIMixin):
-    def start_mock_amapi(self, access_token_value: Optional[str] = None):
+    def start_mock_amapi(self, central_user_db: Optional[AmDB] = None, access_token_value: Optional[str] = None):
         self.start_mock_auth_api(access_token_value=access_token_value)
-
-        self.mocked_users = respx.mock(base_url="http://localhost", assert_all_called=False)
+        self.central_user_db = central_user_db
+        self.mocked_users = respx.mock(base_url="http://localhost/amapi", assert_all_called=False)
         put_users_name_route = self.mocked_users.put(
             url="/users/hubba-bubba/name",
             name="put_users_name_request",
         )
-        put_users_name_route.pass_through()
+        put_users_name_route.mock(side_effect=self._save)
         self.mocked_users.start()
         self.addCleanup(self.mocked_users.stop)  # type: ignore
+
+    def _save(self, request: Request) -> Response:
+        if self.central_user_db is None:
+            raise ValueError("save user side affect was called but self.amdb is None")
+        mock_request = UserUpdateNameRequest.parse_raw(request.content)
+
+        db_user = self.central_user_db.get_user_by_eppn("hubba-bubba")
+        db_user.given_name = mock_request.given_name
+        db_user.surname = mock_request.surname
+        db_user.display_name = mock_request.display_name
+        self.central_user_db.save(user=db_user)
+        return Response(200, text='{"status": "true"}')

src/eduid/common/clients/gnap_client/base.py

@@ -31,7 +31,8 @@ class GNAPClientException(Exception):
 
 
 class GNAPClientAuthData(BaseModel):
-    authn_server_url: str
+    auth_server_url: str
+    auth_server_verify: bool = True
     key_name: str
     client_jwk: JWK
     access: list[Union[str, Access]] = Field(default_factory=list)
@@ -51,7 +52,7 @@ class GNAPBearerTokenMixin(ABC):
 
     @property
     def transaction_uri(self) -> str:
-        return urlappend(self._auth_data.authn_server_url, "transaction")
+        return urlappend(self._auth_data.auth_server_url, "transaction")
 
     def _create_grant_request_jws(self) -> str:
         req = GrantRequest(

src/eduid/common/clients/gnap_client/sync_client.py

@@ -17,7 +17,6 @@ def __init__(self, auth_data: GNAPClientAuthData, **kwargs):
 
         super().__init__(**kwargs)
 
-        self.verify = kwargs.get("verify", True)
         self._auth_data = auth_data
 
     @staticmethod
@@ -34,7 +33,7 @@ def _request_bearer_token(self) -> GrantResponse:
             url=self.transaction_uri,
             content=data,
             headers={"Content-Type": "application/jose+json"},
-            verify=self.verify,
+            verify=self._auth_data.auth_server_verify,
         )
         resp.raise_for_status()
         return GrantResponse.parse_raw(resp.text)

src/eduid/common/rpc/msg_relay.py

@@ -1,3 +1,4 @@
+# -*- coding: utf-8 -*-
 import logging
 from enum import Enum
 from typing import Optional
@@ -121,17 +122,20 @@ def is_deregistered(self) -> bool:
         return bool(self.deregistration_information.cause_code or self.deregistration_information.date)
 
 
-class NavetData(NavetModelConfig):
-    case_information: CaseInformation = Field(alias="CaseInformation")
-    person: Person = Field(alias="Person")
-
-
 # Used to parse data from get_postal_address
 class FullPostalAddress(NavetModelConfig):
     name: Name = Field(default_factory=Name, alias="Name")
     official_address: OfficialAddress = Field(default_factory=OfficialAddress, alias="OfficialAddress")
 
 
+class NavetData(NavetModelConfig):
+    case_information: CaseInformation = Field(alias="CaseInformation")
+    person: Person = Field(alias="Person")
+
+    def get_full_postal_address(self) -> FullPostalAddress:
+        return FullPostalAddress(name=self.person.name, official_address=self.person.postal_addresses)
+
+
 class MsgRelay:
     """
     This is the interface to the RPC task to fetch data from NAVET, and to send SMSs.

src/eduid/common/utils.py

@@ -1,5 +1,11 @@
 __author__ = "lundberg"
 
+from eduid.common.rpc.msg_relay import FullPostalAddress
+from eduid.userdb.user import TUserSubclass
+import logging
+
+logger = logging.getLogger(__name__)
+
 
 def urlappend(base: str, path: str) -> str:
     """
@@ -40,3 +46,33 @@ def removesuffix(s: str, suffix: str) -> str:
         return s[: -len(suffix)]
     else:
         return s[:]
+
+
+def set_user_names_from_official_address(user: TUserSubclass, user_postal_address: FullPostalAddress) -> TUserSubclass:
+    """
+    :param user: Proofing app private userdb user
+    :param user_postal_address: user postal address
+
+    :returns: User object
+    """
+    user.given_name = user_postal_address.name.given_name
+    user.surname = user_postal_address.name.surname
+    if user.given_name is None or user.surname is None:
+        # please mypy
+        raise RuntimeError("No given name or surname found in proofing log user postal address")
+    given_name_marking = user_postal_address.name.given_name_marking
+    user.display_name = f"{user.given_name} {user.surname}"
+    if given_name_marking:
+        _name_index = (int(given_name_marking) // 10) - 1  # ex. "20" -> 1 (second GivenName is real given name)
+        try:
+            _given_name = user.given_name.split()[_name_index]
+            user.display_name = f"{_given_name} {user.surname}"
+        except IndexError:
+            # At least occasionally, we've seen GivenName 'Jan-Erik Martin' with GivenNameMarking 30
+            pass
+    logger.info("User names set from official address")
+    logger.debug(
+        f"{user_postal_address.name} resulted in given_name: {user.given_name}, "
+        f"surname: {user.surname} and display_name: {user.display_name}"
+    )
+    return user

src/eduid/userdb/tests/test_user_cleaner.py

@@ -0,0 +1,14 @@
+from unittest import TestCase
+
+from eduid.userdb.testing import MongoTemporaryInstance
+from eduid.userdb.user_cleaner.cachedb import CacheDB
+from eduid.userdb.user_cleaner.cache import CacheUser
+
+
+class TestUserCleanerCache(TestCase):
+    def setUp(self):
+        self.tmp_db = MongoTemporaryInstance.get_instance()
+        self.user_cleaner_meta_db = CacheDB(db_uri=self.tmp_db.uri, collection="skv_cache")
+
+    def tearDown(self):
+        self.user_cleaner_meta_db._drop_whole_collection()

src/eduid/userdb/user_cleaner/cache.py

@@ -0,0 +1,44 @@
+from typing import Any, Mapping, Optional
+from pydantic import BaseModel, Field
+from datetime import datetime
+from eduid.common.misc.timeutil import utc_now
+
+from eduid.userdb.db import TUserDbDocument
+from eduid.userdb.user import User
+
+
+class CacheUser(BaseModel):
+    eppn: str
+    created_ts: datetime = Field(default_factory=utc_now)
+    next_run_ts: Optional[int] = None
+
+    def to_dict(self) -> TUserDbDocument:
+        """
+        Convert Element to a dict in eduid format, that can be used to reconstruct the
+        Element later.
+        """
+        data = self.dict(exclude_none=True)
+
+        return TUserDbDocument(data)
+
+    @classmethod
+    def from_dict(cls, data: Mapping[str, Any]) -> "CacheUser":
+        """Convert a dict to a Element object."""
+        return cls(**data)
+
+    def next_run_ts_iso8601(self) -> str:
+        """
+        Convert the next_run_ts to ISO8601 format.
+        """
+        if self.next_run_ts is None:
+            return ""
+        return datetime.utcfromtimestamp(self.next_run_ts).strftime("%Y-%m-%d %H:%M:%S")
+
+    def from_user(self, data: User) -> "CacheUser":
+        """
+        Convert a User object to a CacheUser object.
+        """
+        queue_user = CacheUser(
+            eppn=data.eppn,
+        )
+        return queue_user

src/eduid/userdb/user_cleaner/cachedb.py

@@ -0,0 +1,73 @@
+from abc import ABC
+from time import time
+import logging
+from eduid.userdb.db import BaseDB, TUserDbDocument
+from eduid.userdb.user import User
+from eduid.userdb.user_cleaner.cache import CacheUser
+
+logger = logging.getLogger(__name__)
+
+
+class CacheDB(BaseDB):
+    """Database class for the user cleaning database cache."""
+
+    def __init__(self, db_uri: str, collection: str, db_name: str = "eduid_user_cleaner"):
+        super().__init__(db_uri, db_name, collection)
+        indexes = {"unique-eppn": {"key": [("eppn", 1)], "unique": True}}
+
+        self.setup_indexes(indexes)
+
+    def save(self, queue_user: CacheUser) -> bool:
+        """Save a CacheUser object to the database."""
+        if self.exists(queue_user.eppn):
+            logger.debug(f"User {queue_user.eppn} already exists in the queue")
+            return False
+        self._coll.insert_one(queue_user.to_dict())
+        return True
+
+    def exists(self, eppn: str) -> bool:
+        """Check if a user exists in the cache."""
+        return self._coll.count_documents({"eppn": eppn}) > 0
+
+    def get_all(self) -> list[CacheUser]:
+        """Get all users from the cache."""
+        res = self._coll.find({})
+        return [CacheUser.from_dict(data=doc) for doc in res]
+
+    def count(self) -> int:
+        """Count the number of users in the queue."""
+        return self._coll.count_documents({})
+
+    def delete(self, eppn: str) -> None:
+        """delete one user from the cache."""
+        self._coll.delete_one({"eppn": eppn})
+
+    def delete_all(self) -> None:
+        """Delete all users from the cache."""
+        self._coll.delete_many({})
+
+    def populate(
+        self,
+        am_users: list[User],
+        periodicity: int,
+    ) -> None:
+        """Populate cache database with the user from AMDB."""
+
+        cache_size = len(am_users)
+
+        periodicity_in_seconds = 22 * 60 * 60 * periodicity  # strip 2 hours of each day in order to give us some slack
+        time_constant = int(periodicity_in_seconds / cache_size)
+
+        next_run_ts = int(time()) + (60 * 60)  # first process window starts in 1 hour, then the population is done
+        for am_user in am_users:
+            next_run_ts += time_constant
+            queue_user = CacheUser(
+                eppn=am_user.eppn,
+                next_run_ts=next_run_ts,
+            )
+
+            self.save(queue_user)
+
+    def is_empty(self) -> bool:
+        """Check if the cache is empty."""
+        return self.count() == 0