unit tests and more admin functionality

lib/pbench/cli/server/session_management.py

@@ -37,8 +37,13 @@ def user_command_cli(context):
     required=False,
     help="Keycloak realm name",
 )
+@click.option(
+    "--admin_token",
+    required=True,
+    help="Keycloak realm name",
+)
 @common_options
-def get_user_sessions(context, user_id, username, realm):
+def get_user_sessions(context, user_id, username, realm, admin_token):
     try:
         logger = logging.getLogger(__name__)
         config = PbenchServerConfig(context.config)
@@ -47,6 +52,7 @@ def get_user_sessions(context, user_id, username, realm):
             realm_name="Master",
             client_id="admin-cli",
             logger=logger,
+            headers={"Authorization": f"Bearer {admin_token}"},
         )
         if user_id:
             all_user_sessions = keycloak_admin.get_all_user_sessions(
@@ -92,8 +98,13 @@ def get_user_sessions(context, user_id, username, realm):
     required=False,
     help="Keycloak realm name",
 )
+@click.option(
+    "--admin_token",
+    required=True,
+    help="Keycloak realm name",
+)
 @common_options
-def get_client_sessions(context, client_id, client_name, realm):
+def get_client_sessions(context, client_id, client_name, realm, admin_token):
     try:
         logger = logging.getLogger(__name__)
         config = PbenchServerConfig(context.config)
@@ -102,6 +113,7 @@ def get_client_sessions(context, client_id, client_name, realm):
             realm_name="Master",
             client_id="admin-cli",
             logger=logger,
+            headers={"Authorization": f"Bearer {admin_token}"},
         )
         if client_id:
             all_client_sessions = keycloak_admin.get_client_all_sessions(
@@ -128,3 +140,43 @@ def get_client_sessions(context, client_id, client_name, realm):
         click.echo(exc, err=True)
 
     click.get_current_context().exit(rv)
+
+
+@user_command_cli.command()
+@pass_cli_context
+@click.option(
+    "--realm",
+    required=True,
+    help="Keycloak realm name",
+)
+@click.option(
+    "--admin_token",
+    required=True,
+    help="Keycloak realm name",
+)
+@common_options
+def logout_all_realm_sessions(context, realm, admin_token):
+    try:
+        logger = logging.getLogger(__name__)
+        config = PbenchServerConfig(context.config)
+        keycloak_admin = Admin(
+            server_url=config.get("keycloak", "server_url"),
+            realm_name="Master",
+            client_id="admin-cli",
+            logger=logger,
+            headers={"Authorization": f"Bearer {admin_token}"},
+        )
+        status = keycloak_admin.realm_all_sessions_logout(realm_name=realm)
+        click.echo(status)
+        rv = 0
+    except BadConfig as exc:
+        rv = 2
+        click.echo(exc, err=True)
+    except KeycloakConnectionError as exc:
+        rv = 3
+        click.echo(exc, err=True)
+    except Exception as exc:
+        rv = 1
+        click.echo(exc, err=True)
+
+    click.get_current_context().exit(rv)

lib/pbench/server/auth/__init__.py

@@ -1,9 +1,11 @@
 from http import HTTPStatus
-from typing import Dict, List, Union
+from typing import Dict, List, Union, Optional
 from urllib.parse import urljoin
 
+import logging
 import jwt
 import requests
+from requests.structures import CaseInsensitiveDict
 
 from pbench.server import JSON
 from pbench.server.auth.auth_provider_urls import (
@@ -33,18 +35,18 @@ def __init__(
         server_url: str,
         realm_name: str,
         client_id: str,
-        logger,
+        logger: logging.Logger,
         client_secret_key: str = None,
         verify: bool = True,
-        headers: Dict = None,
+        headers: Optional[Dict[str, str]] = None,
         timeout: int = 60,
     ):
         self.server_url = server_url
         self.client_id = client_id
         self.client_secret_key = client_secret_key
         self.realm_name = realm_name
         self.logger = logger
-        self.headers = headers if headers is not None else dict()
+        self.headers = headers if headers is not None else CaseInsensitiveDict()
         self.verify = verify
         self.timeout = timeout
         self.connection = requests.session()
@@ -67,7 +69,7 @@ def del_param_headers(self, key: str):
         """Remove a specific header parameter.
         :param key: Key of the header parameters.
         """
-        self.headers.pop(key, None)
+        del self.headers[key]
 
     def get_well_known(self) -> JSON:
         """Returns the well-known configuration endpoints as a JSON.
@@ -188,14 +190,15 @@ def token_introspect_offline(
             token, key, algorithms=algorithms, audience=audience, **kwargs
         )
 
-    def get_userinfo(self, token: str) -> JSON:
+    def get_userinfo(self, token: str = None) -> JSON:
         """
         The userinfo endpoint returns standard claims about the authenticated user,
         and is protected by a bearer token.
         http://openid.net/specs/openid-connect-core-1_0.html#UserInfo
         """
 
-        self.add_header_param("Authorization", f"Bearer {token}")
+        if token:
+            self.add_header_param("Authorization", f"Bearer {token}")
         params_path = {"realm-name": self.realm_name}
 
         return self._get(URL_USERINFO.format(**params_path)).json()

lib/pbench/server/auth/auth_provider_urls.py

@@ -8,12 +8,17 @@
 
 # Admin URIs
 URL_ADMIN_SERVER_INFO = "admin/serverinfo"
+URL_ADMIN_REALM_SESSIONS_LOGOUT = "admin/realms/{realm-name}/logout-all"
 URL_ADMIN_KEYS = "admin/realms/{realm-name}/keys"
 URL_ADMIN_CLIENTS = "admin/realms/{realm-name}/clients"
 URL_ADMIN_CLIENT = URL_ADMIN_CLIENTS + "/{id}"
+URL_ADMIN_CLIENT_SESSIONS_COUNT = URL_ADMIN_CLIENT + "/session-count"
 URL_ADMIN_CLIENT_ALL_SESSIONS = URL_ADMIN_CLIENT + "/user-sessions"
 URL_ADMIN_CLIENT_ROLES = URL_ADMIN_CLIENT + "/roles"
+URL_ADMIN_CLIENT_ROLE = URL_ADMIN_CLIENT + "/roles/{role-name}"
 URL_ADMIN_USERS = "admin/realms/{realm-name}/users"
+URL_ADMIN_USER_LOGOUT = URL_ADMIN_USERS + "/{id}/logout"
+URL_ADMIN_USERS_COUNT = "admin/realms/{realm-name}/users/count"
 URL_ADMIN_USER = "admin/realms/{realm-name}/users/{id}"
 URL_ADMIN_GET_SESSIONS = "admin/realms/{realm-name}/users/{id}/sessions"
 URL_ADMIN_REALM_ROLES = "admin/realms/{realm-name}/roles"