chobotnice group sync

contracts/templates/contracts/includes/contract_list.html

-<ul class="grid grid-cols-2 gap-4 flex">
+<ul class="grid grid-cols-1 md:grid-cols-2 gap-4 flex">
     {% for contract in page %}
         <li class="card elevation-10{% if not contract.is_public %} bg-red-100{% endif %}">
             <div class="card__body p-5">

env.example

@@ -23,3 +23,5 @@ DEFAULT_STAFF_GROUPS="sso_cen:f,sso_cen:neverejni,sso_cen:smlouvy_ao,sso_cen:sml
 
 NASTENKA_API_URL=http://localhost:8009/contracts/api/notices
 NASTENKA_API_TOKEN=cirno
+
+CHOBOTNICE_API_URL="https://chobotnice.pirati.cz/graphql/"

oidc/auth.py

 import logging
 import typing
 
+import gql
 import jwt
 from django.conf import settings
 from django.contrib.auth.models import Group
+from django_http_exceptions import HTTPExceptions
+from gql.transport.exceptions import TransportQueryError
+from gql.transport.requests import RequestsHTTPTransport
 from pirates.auth import PiratesOIDCAuthenticationBackend
 
 logging.basicConfig(level=logging.DEBUG)
@@ -11,16 +15,13 @@ logging.basicConfig(level=logging.DEBUG)
 
 class RegistryOIDCAuthenticationBackend(PiratesOIDCAuthenticationBackend):
     def _assign_new_user_groups(
-        self, user, access_token: dict, user_groups: typing.Union[None, list] = None
+        self, user, new_user_groups: list, existing_user_groups=None
     ) -> None:
-        if user_groups is None:
-            user_groups = user.groups.all()
+        if existing_user_groups is None:
+            existing_user_groups = user.groups.all()
 
-        for group in access_token["groups"]:
-            if group.startswith("_"):  # Ignore internal Keycloak groups
-                continue
-
-            group_name = f"sso_{group}"
+        for group in new_user_groups:
+            group_name = f"chobo_{group}"
 
             group = Group.objects.filter(name=group_name)
 
@@ -30,19 +31,62 @@ class RegistryOIDCAuthenticationBackend(PiratesOIDCAuthenticationBackend):
             else:
                 group = group[0]
 
-            if group not in user_groups:
+            if group not in existing_user_groups:
                 user.groups.add(group)
 
+        user.save()
+
     def _remove_old_user_groups(
-        self, user, access_token: dict, user_groups: typing.Union[None, list] = None
+        self, user, new_user_groups: list, existing_user_groups=None
     ) -> None:
-        if user_groups is None:
-            user_groups = user.groups.all()
+        if existing_user_groups is None:
+            existing_user_groups = user.groups.all()
 
-        for group in user_groups:
-            if group.name.replace("sso_", "") not in access_token["groups"]:
+        for group in existing_user_groups:
+            if group.name.replace("chobo_", "") not in new_user_groups:
                 user.groups.remove(group)
 
+    def get_chobotnice_groups(self, access_token):
+        transport = RequestsHTTPTransport(url=settings.CHOBOTNICE_API_URL)
+        client = gql.Client(
+            transport=transport,
+            fetch_schema_from_transport=True,
+        )
+
+        query = gql.gql(
+            f"""
+                {{
+                    allPeople(
+                        filters: {{keycloakId: {{exact: "{access_token['sub']}"}}}}
+                    ) {{
+                        edges {{
+                            node {{
+                                groupMemberships {{
+                                    group {{
+                                        shortcut
+                                    }}
+                                }}
+                            }}
+                        }}
+                    }}
+                }}
+            """
+        )
+
+        try:
+            result = client.execute(query)
+        except TransportQueryError:
+            # rv_gid was not found
+            raise HTTPExceptions.BAD_REQUEST
+
+        groups = []
+
+        for person in result["allPeople"]["edges"]:
+            for group_membership in person["node"]["groupMemberships"]:
+                groups.append(group_membership["group"]["shortcut"])
+
+        return groups
+
     def get_or_create_user(self, access_token, id_token, payload):
         user = super().get_or_create_user(access_token, id_token, payload)
 
@@ -54,13 +98,18 @@ class RegistryOIDCAuthenticationBackend(PiratesOIDCAuthenticationBackend):
         )
 
         user.preferred_username = decoded_access_token["preferred_username"]
-        user_groups = user.groups.all()
+        existing_user_groups = user.groups.all()
+        new_user_groups = self.get_chobotnice_groups(decoded_access_token)
 
         self._remove_old_user_groups(
-            user, decoded_access_token, user_groups=user_groups
+            user,
+            new_user_groups=new_user_groups,
+            existing_user_groups=existing_user_groups,
         )
         self._assign_new_user_groups(
-            user, decoded_access_token, user_groups=user_groups
+            user,
+            new_user_groups=new_user_groups,
+            existing_user_groups=existing_user_groups,
         )
 
         user.update_group_based_admin()

registry/settings/base.py

@@ -160,6 +160,9 @@ OIDC_OP_AUTHORIZATION_ENDPOINT = OIDC_RP_REALM_URL + "protocol/openid-connect/au
 OIDC_OP_TOKEN_ENDPOINT = OIDC_RP_REALM_URL + "protocol/openid-connect/token"
 OIDC_OP_USER_ENDPOINT = OIDC_RP_REALM_URL + "protocol/openid-connect/userinfo"
 
+# Chobotnice - group source
+CHOBOTNICE_API_URL = env.str("CHOBOTNICE_API_URL")
+
 
 ## Internationalization
 # https://docs.djangoproject.com/en/4.0/topics/i18n/

requirements/base.txt

@@ -9,8 +9,9 @@ django-dbsettings==1.3.0
 django-downloadview==2.3.0
 django-nested-admin==4.0.2
 django-ordered-model==3.7.1
-psycopg2-binary==2.9.5
 django-webpack-loader==1.8.0
+gql[requests]==3.4.0
+psycopg2-binary==2.9.5
 pirates==0.6.0
 django-markdownx==4.0.0b1
 django-environ==0.9.0