[#2932] Get vestigingsnummer from OIDC claim on login & store in session

src/eherkenning/mock/backends.py

@@ -22,6 +22,8 @@ class eHerkenningBackend(BaseBackend):
         }
     )
 
+    # TODO: update mock to test retrieval/storage of vestigingsnummer
+
     def get_or_create_user(self, request, kvk):
         created = False
         try:

src/open_inwoner/accounts/backends.py

@@ -5,15 +5,20 @@
 from django.contrib.auth import get_user_model
 from django.contrib.auth.backends import ModelBackend
 from django.contrib.auth.hashers import check_password
+from django.contrib.auth.models import AbstractUser
+from django.core.exceptions import SuspiciousOperation
 from django.urls import reverse, reverse_lazy
 
 from axes.backends import AxesBackend
 from digid_eherkenning.oidc.backends import BaseBackend
+from digid_eherkenning.oidc.claims import process_claims
 from mozilla_django_oidc_db.backends import OIDCAuthenticationBackend
 from mozilla_django_oidc_db.config import dynamic_setting
+from mozilla_django_oidc_db.typing import JSONObject
 from oath import accept_totp
 
 from open_inwoner.configurations.models import SiteConfiguration
+from open_inwoner.kvk.branches import KVK_BRANCH_SESSION_VARIABLE
 from open_inwoner.utils.hash import generate_email_from_string
 
 from .choices import LoginTypeChoices
@@ -186,3 +191,49 @@ def create_user(self, claims):
         )
 
         return user
+
+    def update_user(self, user: AbstractUser, claims: JSONObject):
+        processed_claims = process_claims(claims, config=self.config_class)
+        if vestigingsnummer := processed_claims.get("vestigingsnummer", None):
+            self.request.session["vestigingsnummer"] = vestigingsnummer
+        return user
+
+    def get_or_create_user(self, access_token, id_token, payload):
+        """Override to store vestigingsnummer (if already present) in session"""
+
+        user_info = self.get_userinfo(access_token, id_token, payload)
+
+        claims_verified = self.verify_claims(user_info)
+        if not claims_verified:
+            msg = "Claims verification failed"
+            raise SuspiciousOperation(msg)
+
+        claims_processed = process_claims(claims_verified, config=self.config_class)
+
+        # email based filtering
+        users = self.filter_users_by_claims(user_info)
+
+        if len(users) == 1:
+            self.request.session[KVK_BRANCH_SESSION_VARIABLE] = claims_processed.get(
+                "vestigingsnummer", None
+            )
+            self.request.session.save()
+            return self.update_user(users[0], user_info)
+        elif len(users) > 1:
+            # In the rare case that two user accounts have the same email address,
+            # bail. Randomly selecting one seems really wrong.
+            msg = "Multiple users returned"
+            raise SuspiciousOperation(msg)
+        elif self.get_settings("OIDC_CREATE_USER", True):
+            user = self.create_user(user_info)
+            self.request.session[KVK_BRANCH_SESSION_VARIABLE] = claims_processed.get(
+                "vestigingsnummer", None
+            )
+            self.request.session.save()
+            return user
+        else:
+            logger.debug(
+                "Login failed: No user with %s found, and " "OIDC_CREATE_USER is False",
+                self.describe_user_by_claims(user_info),
+            )
+            return None

src/open_inwoner/kvk/views.py

@@ -59,6 +59,10 @@ def get(self, request, *args, **kwargs):
             return HttpResponse(_("Unauthorized"), status=401)
 
         redirect = self.get_redirect()
+
+        if request.session.get(KVK_BRANCH_SESSION_VARIABLE, None):
+            return HttpResponseRedirect(redirect)
+
         context = super().get_context_data()
 
         form = context["form"]