Add roles/groups of jwt token

tdp_core/security/store/alb_security_store.py

@@ -11,20 +11,39 @@
 
 
 class ALBSecurityStore(BaseStore):
-    def __init__(self, cookie_name: Optional[str], signout_url: Optional[str]):
+    def __init__(
+        self, cookie_name: Optional[str], signout_url: Optional[str], token_user_attr: Optional[str], token_roles_attr: Optional[str]
+    ):
         self.cookie_name = cookie_name
-        self.signout_url: Optional[str] = signout_url
+        self.signout_url = signout_url
+        self.token_user_attr = token_user_attr
+        self.token_roles_attr = token_roles_attr
 
     def load_from_request(self, req):
         if "X-Amzn-Oidc-Identity" in req.headers and "X-Amzn-Oidc-Accesstoken" in req.headers and "X-Amzn-Oidc-Data" in req.headers:
             try:
+                roles = []
                 # Get token data from header
+                _log.debug(f"headers: {req.headers}")
                 encoded = req.headers["X-Amzn-Oidc-Data"]
+                _log.debug(f"X-Amzn-Oidc-Data: {encoded}")
+                _log.debug(f"X-Amzn-Oidc-Accesstoken: {req.headers['X-Amzn-Oidc-Accesstoken']}")
+                _log.debug(f"X-Amzn-Oidc-Identity: {req.headers['X-Amzn-Oidc-Identity']}")
                 # Try to decode the oidc data jwt
-                user = jwt.decode(encoded, options={"verify_signature": False})
+                user_data = jwt.decode(encoded, options={"verify_signature": False})
+                _log.debug(f"user data: {user_data}")
                 # Create new user from given attributes
-                email = user["email"]
-                return User(id=email, roles=[])
+                user = user_data[self.token_user_attr]
+                _log.debug("user: %s", user)
+                if self.token_roles_attr:
+                    roles = user_data[self.token_roles_attr]
+                _log.debug("roletype: %s", type(roles))
+                if not roles:
+                    roles = []
+                elif type(roles) == str:
+                    roles = [roles]
+                _log.debug("roles: %s", roles)
+                return User(id=user, roles=roles)
             except Exception:
                 _log.exception("Error in load_from_request")
                 return None
@@ -54,6 +73,8 @@ def create():
         return ALBSecurityStore(
             manager.settings.tdp_core.security.store.alb_security_store.cookie_name,
             manager.settings.tdp_core.security.store.alb_security_store.signout_url,
+            manager.settings.tdp_core.security.store.alb_security_store.token_user_attr,
+            manager.settings.tdp_core.security.store.alb_security_store.token_roles_attr,
         )
 
     return None

tdp_core/settings/model.py

@@ -31,6 +31,8 @@ class AlbSecurityStoreSettings(BaseModel):
     enable: bool = False
     cookie_name: Optional[str] = None
     signout_url: Optional[str] = None
+    token_user_attr: Optional[str] = "email"
+    token_roles_attr: Optional[str] = None
 
 
 class NoSecurityStoreSettings(BaseModel):