Feature/ogc 3157 current user none crash

src/onegov/core/security/identity_policy.py

@@ -16,7 +16,7 @@ class IdentityPolicy:
 
     """
 
-    required_keys = {'userid', 'groupids', 'role', 'application_id'}
+    required_keys = {'userid', 'uid', 'groupids', 'role', 'application_id'}
 
     def identify(self, request: CoreRequest) -> Identity | None:
         try:

src/onegov/org/forms/user.py

@@ -202,8 +202,7 @@ def process(
     def populate_obj(self, obj: User) -> None:  # type: ignore[override]
         assert self.new_username.data is not None
         request = self.request
-        obj.logout_all_sessions(request.app)
-        obj.username = self.new_username.data
+        obj.change_username(self.new_username.data, request.app)
         # Run application-specific callback
         request.app.settings.user.change_username_callback(obj, request)
 

src/onegov/org/request.py

@@ -1,16 +1,18 @@
 from __future__ import annotations
 
+
 from functools import cached_property
 from onegov.core.custom import msgpack
 from onegov.core.orm import orm_cached
-from onegov.core.request import CoreRequest
+from onegov.core.request import CoreRequest, is_logged_in
 from onegov.core.security import Private
 from onegov.core.utils import normalize_for_url
 from onegov.org.models import News, TANAccessCollection, Topic
 from onegov.page import Page, PageCollection
 from onegov.user import User
 from sedate import utcnow
 from sqlalchemy import inspect
+from uuid import UUID
 from sqlalchemy.orm import noload
 
 
@@ -103,22 +105,32 @@ def is_member(self) -> bool:
 
     @property
     def current_username(self) -> str | None:
+        cached = getattr(self, '_current_user', None)
+        if cached is not None:
+            return cached.username
+
         return self.identity.userid if self.identity else None
 
     # NOTE: Internal cache, don't use directly!
     _current_user: User | None
 
     @property
     def current_user(self) -> User | None:
-        if not self.identity:
+        if not is_logged_in(self.identity):
             return None
 
         if not hasattr(self, '_current_user'):
             self._current_user = (
                 self.session.query(User)
-                .filter_by(username=self.identity.userid)
+                .filter(User.id == UUID(self.identity.uid))
                 .first()
             )
+            if (
+                self._current_user is not None
+                and self.identity.userid != self._current_user.username
+            ):
+                # stale username in session — fix it for subsequent requests
+                self.browser_session['userid'] = self._current_user.username
             return self._current_user
 
         if self._current_user is None:

src/onegov/translator_directory/generate_docx.py

@@ -188,8 +188,10 @@ def signature_for_mail_templates(
     first and last name of the user. """
 
     assert request.current_user is not None
-    assert request.current_user.realname is not None
-    first_name, last_name = request.current_user.realname.split(' ')
+    name_parts = (request.current_user.realname or '').strip().split(None, 1)
+    if len(name_parts) != 2 or not name_parts[0] or not name_parts[1]:
+        return None
+    first_name, last_name = name_parts
     query = GeneralFileCollection(request.session).query().filter(
         and_(
             GeneralFile.name.like('Unterschrift%'),

src/onegov/user/auth/core.py

@@ -272,11 +272,6 @@ def as_identity(self, user: User) -> Identity:
         """ Returns the morepath identity of the given user. """
 
         return self.identity_class(
-            # FIXME: We should consider switching to user.id.hex
-            #        for userid and provide username in a separate field
-            #        instead, but this was the less intrusive step
-            #        in the meantime. We use identity.userid in quite
-            #        a few places after all.
             userid=user.username,
             uid=user.id.hex,
             groupids=frozenset(group.id.hex for group in user.groups),
@@ -286,9 +281,11 @@ def as_identity(self, user: User) -> Identity:
 
     def by_identity(self, identity: Identity | NoIdentity) -> User | None:
         """ Returns the user record of the given identity. """
+
         if identity.userid is None:
             return None
-        return self.users.by_username(identity.userid)
+
+        return self.users.by_id(identity.uid)
 
     def login_to(
         self,

src/onegov/user/auth/provider.py

@@ -324,7 +324,7 @@ def ensure_user(
         if users.by_username(username) is not None:
             log.error(f'Cannot rename user {user.username} to {username}')
         else:
-            user.username = username
+            user.change_username(username, app)
             app.settings.user.change_username_callback(user, request)
 
     # update the role even if the user exists already

src/onegov/user/cli.py

@@ -282,8 +282,7 @@ def change(request: CoreRequest, app: Framework) -> None:
         if users.exists(new_username):
             abort(f'{new_username} already exists')
 
-        user.logout_all_sessions(request.app)
-        user.username = new_username
+        user.change_username(new_username, request.app)
 
         # Run application-specific callback
         app.settings.user.change_username_callback(user, request)

src/onegov/user/models/user.py

@@ -362,3 +362,8 @@ def logout_all_sessions(self, app: Framework) -> int:
         self.cleanup_sessions(app)
 
         return count
+
+    def change_username(self, new_username: str, app: Framework) -> None:
+        """ Changes the username, logging out all active sessions first. """
+        self.logout_all_sessions(app)
+        self.username = new_username

tests/onegov/org/test_views_user.py

@@ -115,6 +115,46 @@ def test_change_username(client: Client) -> None:
     assert 'editor2@example.org' in users_page
 
 
+def test_change_username_invalidates_sessions(client: Client) -> None:
+    editor = client.spawn()
+    editor.login_editor()
+    assert editor.get('/userprofile').status_code == 200
+
+    # Admin renames the editor via change_username (logs out all sessions)
+    transaction.begin()
+    users = UserCollection(client.app.session())
+    user = users.by_username('editor@example.org')
+    assert user is not None
+    user.change_username('editor.renamed@example.org', client.app)
+    transaction.commit()
+    close_all_sessions()
+
+    # The editor's existing session should now be invalid
+    assert editor.get('/userprofile', expect_errors=True).status_code == 403
+
+
+def test_session_survives_username_rename(client: Client) -> None:
+    editor = client.spawn()
+    editor.login_editor()
+    assert editor.get('/userprofile').status_code == 200
+
+    # Rename the editor directly in the DB, bypassing change_username.
+    # This simulates an LDAP sync or any path that updates the username
+    # without invalidating the session.
+    transaction.begin()
+    users = UserCollection(client.app.session())
+    user = users.by_username('editor@example.org')
+    assert user is not None
+    user.username = 'editor.renamed@example.org'
+    transaction.commit()
+    close_all_sessions()
+
+    # The existing session should still work via primary key lookup.
+    # The stale userid in the session is also updated on this request so
+    # that code reading identity.userid directly sees the correct username.
+    assert editor.get('/userprofile').status_code == 200
+
+
 def test_user_source(client: Client) -> None:
     client.login_admin()
 

tests/onegov/user/test_models.py

@@ -176,6 +176,25 @@ def test_user_logout_all_sessions() -> None:
             assert call(None, 'yyy') in forget.mock_calls
 
 
+def test_user_change_username() -> None:
+    user = User(username='old@example.org')
+
+    with (
+        patch('onegov.user.models.user.remembered') as remembered,
+        patch('onegov.user.models.user.forget') as forget
+    ):
+        remembered.return_value = True
+
+        user.save_current_session(DummyRequest('sess1'))  # type: ignore[arg-type]
+        assert user.sessions is not None
+        assert 'sess1' in user.sessions
+
+        user.change_username('new@example.org', DummyRequest('zzz').app)  # type: ignore[arg-type]
+
+        assert call(None, 'sess1') in forget.mock_calls
+        assert user.username == 'new@example.org'
+
+
 def test_user_cleanup_sessions() -> None:
     user = User()
     assert not user.sessions