34987 model sync user with token

python/common/business-registry-model/pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "business-model"
-version = "3.4.15"
+version = "3.4.16"
 description = ""
 authors = [
     {name = "thor",email = "1042854+thorwolpert@users.noreply.github.com"}

python/common/business-registry-model/src/business_model/models/user.py

@@ -120,14 +120,45 @@ def create_from_jwt_token(cls, token: dict):
             return user
         return None
 
+    @classmethod
+    def _sync_from_jwt_token(cls, user, token: dict):
+        """Refresh the user record with the values found in the JWT token."""
+        changed = False
+
+        username = token.get(current_app.config.get('JWT_OIDC_USERNAME'), None)
+        if username is not None and username != user.username:
+            user.username = username
+            changed = True
+
+        firstname = token.get(current_app.config.get('JWT_OIDC_FIRSTNAME'), None)
+        lastname = token.get(current_app.config.get('JWT_OIDC_LASTNAME'), None)
+        if firstname is not None and lastname is not None:
+            token_name = ' '.join(f'{firstname} {lastname}'.upper().split())
+            record_name = ' '.join(
+                ' '.join(filter(None, [user.firstname, user.middlename, user.lastname])).upper().split())
+            if token_name != record_name:
+                user.firstname = firstname
+                # Current Token values do not have middlename (middle name is combined into firstname instead)
+                user.middlename = None
+                user.lastname = lastname
+                changed = True
+
+        if changed:
+            current_app.logger.debug(f'Updating user from JWT:{token}; User:{user}')
+            db.session.add(user)
+            db.session.commit()
+
+        return user
+
     @classmethod
     def get_or_create_user_by_jwt(cls, jwt_oidc_token):
         """Return a valid user for audit tracking purposes."""
         # GET existing or CREATE new user based on the JWT info
         try:
-            user = User.find_by_jwt_token(jwt_oidc_token)
             current_app.logger.debug(f'finding user: {jwt_oidc_token}')
-            if not user:
+            if user := User.find_by_jwt_token(jwt_oidc_token):
+                user = User._sync_from_jwt_token(user, jwt_oidc_token)  # pylint: disable=protected-access
+            else:
                 current_app.logger.debug(f'didnt find user, attempting to create new user:{jwt_oidc_token}')
                 user = User.create_from_jwt_token(jwt_oidc_token)
 

python/common/business-registry-model/tests/models/test_user.py

@@ -28,6 +28,14 @@
 def create_sub():
     return base64.urlsafe_b64encode(uuid.uuid4().bytes).decode().replace('=', '')
 
+
+def _set_jwt_oidc_claim_config(app, monkeypatch):
+    """Configure the JWT claim names so token claims map onto the User attributes."""
+    monkeypatch.setitem(app.config, 'JWT_OIDC_USERNAME', 'username')
+    monkeypatch.setitem(app.config, 'JWT_OIDC_FIRSTNAME', 'firstname')
+    monkeypatch.setitem(app.config, 'JWT_OIDC_LASTNAME', 'lastname')
+
+
 def test_user(session):
     """Assert that a User can be stored in the service.
 
@@ -103,6 +111,136 @@ def test_get_or_create_user_by_jwt_invlaid_jwt(session):
     assert excinfo.value.error == 'unable_to_get_or_create_user'
 
 
+def test_get_or_create_user_by_jwt_creates_new_user_with_token_claims(session, app, monkeypatch):
+    """Assert that a brand new user is created and populated from the JWT claims."""
+    _set_jwt_oidc_claim_config(app, monkeypatch)
+    idp_userid = create_sub()
+    token = {'username': 'newusername',
+             'firstname': 'New',
+             'lastname': 'User',
+             'iss': 'iss',
+             'sub': create_sub(),
+             'idp_userid': idp_userid,
+             'loginSource': 'BCSC'
+             }
+
+    u = User.get_or_create_user_by_jwt(token)
+
+    assert u.id is not None
+    assert u.username == 'newusername'
+    assert u.firstname == 'New'
+    assert u.lastname == 'User'
+
+
+@pytest.mark.parametrize('test_name, attribute', [
+    ('firstname changed', 'firstname'),
+    ('lastname changed', 'lastname'),
+    ('username changed', 'username'),
+])
+def test_get_or_create_user_by_jwt_syncs_changed_claim(session, app, monkeypatch, test_name, attribute):
+    """Assert that an existing user's stored attribute is refreshed when the JWT claim has changed."""
+    _set_jwt_oidc_claim_config(app, monkeypatch)
+    idp_userid = create_sub()
+    user = User(username='username', firstname='firstname', lastname='lastname',
+                sub=create_sub(), iss='iss', idp_userid=idp_userid, login_source='IDIR')
+    session.add(user)
+    session.commit()
+    user_id = user.id
+
+    token = {'username': 'username', 'firstname': 'firstname', 'lastname': 'lastname',
+             'iss': 'iss', 'sub': create_sub(), 'idp_userid': idp_userid, 'loginSource': 'IDIR'}
+    token[attribute] = 'updated_value'
+
+    u = User.get_or_create_user_by_jwt(token)
+
+    assert u.id == user_id
+    assert getattr(u, attribute) == 'updated_value'
+
+
+def test_get_or_create_user_by_jwt_leaves_matching_claims_unchanged(session, app, monkeypatch):
+    """Assert that an existing user's attributes are left unchanged when the JWT claims already match."""
+    _set_jwt_oidc_claim_config(app, monkeypatch)
+    idp_userid = create_sub()
+    user = User(username='username', firstname='firstname', lastname='lastname',
+                sub=create_sub(), iss='iss', idp_userid=idp_userid, login_source='IDIR')
+    session.add(user)
+    session.commit()
+    user_id = user.id
+
+    token = {'username': 'username', 'firstname': 'firstname', 'lastname': 'lastname',
+             'iss': 'iss', 'sub': create_sub(), 'idp_userid': idp_userid, 'loginSource': 'IDIR'}
+
+    u = User.get_or_create_user_by_jwt(token)
+
+    assert u.id == user_id
+    assert u.username == 'username'
+    assert u.firstname == 'firstname'
+    assert u.lastname == 'lastname'
+
+
+def test_get_or_create_user_by_jwt_preserves_stored_values_when_claims_missing(session, app, monkeypatch):
+    """Assert that stored name values survive a token with no name claims (eg. a service account)."""
+    _set_jwt_oidc_claim_config(app, monkeypatch)
+    idp_userid = create_sub()
+    user = User(username='username', firstname='firstname', lastname='lastname',
+                sub=create_sub(), iss='iss', idp_userid=idp_userid, login_source='IDIR')
+    session.add(user)
+    session.commit()
+    user_id = user.id
+
+    # service-account style token: no username/firstname/lastname claims present
+    token = {'iss': 'iss', 'sub': create_sub(), 'idp_userid': idp_userid, 'loginSource': 'system'}
+
+    u = User.get_or_create_user_by_jwt(token)
+
+    assert u.id == user_id
+    assert u.username == 'username'
+    assert u.firstname == 'firstname'
+    assert u.lastname == 'lastname'
+
+
+def test_get_or_create_user_by_jwt_split_name_is_not_stale(session, app, monkeypatch):
+    """Assert that a record splitting the same name differently is left untouched."""
+    _set_jwt_oidc_claim_config(app, monkeypatch)
+    idp_userid = create_sub()
+    user = User(username='username', firstname='Joe', middlename='P', lastname='Swanson',
+                sub=create_sub(), iss='iss', idp_userid=idp_userid, login_source='BCSC')
+    session.add(user)
+    session.commit()
+    user_id = user.id
+
+    token = {'username': 'username', 'firstname': 'Joe P', 'lastname': 'Swanson',
+             'iss': 'iss', 'sub': create_sub(), 'idp_userid': idp_userid, 'loginSource': 'BCSC'}
+
+    u = User.get_or_create_user_by_jwt(token)
+
+    assert u.id == user_id
+    assert u.firstname == 'Joe'
+    assert u.middlename == 'P'
+    assert u.lastname == 'Swanson'
+
+
+def test_get_or_create_user_by_jwt_name_change_clears_middlename(session, app, monkeypatch):
+    """Assert that a genuine name change replaces the whole stored name, middlename included."""
+    _set_jwt_oidc_claim_config(app, monkeypatch)
+    idp_userid = create_sub()
+    user = User(username='username', firstname='Joe', middlename='P', lastname='Swanson',
+                sub=create_sub(), iss='iss', idp_userid=idp_userid, login_source='BCSC')
+    session.add(user)
+    session.commit()
+    user_id = user.id
+
+    token = {'username': 'username', 'firstname': 'Joseph P', 'lastname': 'Swanson',
+             'iss': 'iss', 'sub': create_sub(), 'idp_userid': idp_userid, 'loginSource': 'BCSC'}
+
+    u = User.get_or_create_user_by_jwt(token)
+
+    assert u.id == user_id
+    assert u.firstname == 'Joseph P'
+    assert u.middlename is None
+    assert u.lastname == 'Swanson'
+
+
 def test_create_from_jwt_token_no_token(session):
     """Assert User is not created from an empty token."""
     token = None