activate dockerized direct access grant on simulator

Author: felixevers

docker-compose.dev.yml

@@ -109,6 +109,9 @@ services:
       KEYCLOAK_URL: "http://keycloak:8080"
       API_URL: "http://backend:80/graphql"
       REALM: "tasks"
+      USE_DIRECT_GRANT: "true"
+      USERNAME: "test"
+      PASSWORD: "test"
       CLIENT_ID: "tasks-web"
     depends_on:
       - backend

docker-compose.yml

@@ -114,6 +114,9 @@ services:
       KEYCLOAK_URL: "http://keycloak:8080/keycloak"
       API_URL: "http://backend:80/graphql"
       REALM: "tasks"
+      USE_DIRECT_GRANT: "true"
+      USERNAME: "test"
+      PASSWORD: "test"
       CLIENT_ID: "tasks-web"
     depends_on:
       - backend

keycloak/tasks.json

@@ -682,7 +682,7 @@
     "consentRequired" : false,
     "standardFlowEnabled" : true,
     "implicitFlowEnabled" : false,
-    "directAccessGrantsEnabled" : false,
+    "directAccessGrantsEnabled" : true,
     "serviceAccountsEnabled" : false,
     "publicClient" : true,
     "frontchannelLogout" : true,

simulator/Dockerfile

@@ -16,11 +16,11 @@ FROM python:${PY_VERSION}-alpine
 
 ENV PYTHONDONTWRITEBYTECODE=1
 ENV PYTHONUNBUFFERED=1
+ENV USE_DIRECT_GRANT=true
 
 COPY --from=builder /build/venv /usr/local/
 COPY . /app
 
 WORKDIR /app
 
-CMD ["python", "-m", "simulator"]
-
+CMD ["python", "main.py"]

simulator/README.md

@@ -8,7 +8,7 @@ This script is designed solely for development and testing environments to mimic
 
 ## Features
 
-- **Authentication**: Implements OpenID Connect Authorization Code flow with a local callback server.
+- **Authentication**: Supports both interactive (browser-based) and non-interactive (direct grant) authentication flows.
 - **Location Management**: Displays hospital structure and manages location hierarchies.
 - **Patient Simulation**:
   - Creates new patients in waiting room or admits them directly.
@@ -33,13 +33,25 @@ This script is designed solely for development and testing environments to mimic
 KEYCLOAK_URL=http://localhost:8080
 API_URL=http://localhost:8000/graphql
 REALM=tasks
-CLIENT_ID=tasks-web
+USE_DIRECT_GRANT=false  # Set to "true" for non-interactive authentication
+CLIENT_ID=tasks-web  # Automatically set to "admin-cli" when USE_DIRECT_GRANT=true
+CLIENT_SECRET=  # Optional, required if client is confidential
+USERNAME=  # Required when USE_DIRECT_GRANT=true
+PASSWORD=  # Required when USE_DIRECT_GRANT=true
 INFLUXDB_URL=http://localhost:8086
 INFLUXDB_TOKEN=tasks-token-secret
 INFLUXDB_ORG=tasks
 INFLUXDB_BUCKET=audit
 ```
 
+### Authentication Modes
+
+The simulator supports two authentication modes:
+
+1. **Interactive (Default)**: Browser-based OAuth2 Authorization Code flow with a local callback server. This is the default mode and requires no additional configuration.
+
+2. **Non-Interactive**: Set `USE_DIRECT_GRANT=true` along with `USERNAME` and `PASSWORD` environment variables. The simulator will authenticate using the direct grant (resource owner password credentials) flow without opening a browser. This mode is automatically enabled in Docker containers.
+
 ## Usage
 
 ### Local Development
@@ -93,7 +105,9 @@ The simulator is split into multiple modules:
 
 ## How It Works
 
-1. Authenticates with Keycloak using OAuth2 Authorization Code flow
+1. Authenticates with Keycloak using either:
+   - Interactive: OAuth2 Authorization Code flow (default) - opens browser
+   - Non-Interactive: Direct Grant flow (if USE_DIRECT_GRANT=true and USERNAME/PASSWORD are set) - no browser required
 2. Loads current state (locations, patients, tasks, users)
 3. Displays the location structure hierarchy
 4. Creates initial patients (some in waiting room, some admitted)

simulator/authentication.py

@@ -1,15 +1,20 @@
+import threading
+import time
 import webbrowser
 from http.server import BaseHTTPRequestHandler, HTTPServer
 from typing import Any
 from urllib.parse import parse_qs, urlparse
-import threading
+
 import requests
 from config import (
     CALLBACK_PORT,
     CLIENT_ID,
+    CLIENT_SECRET,
     KEYCLOAK_URL,
+    PASSWORD,
     REALM,
     REDIRECT_URI,
+    USERNAME,
     logger,
 )
 
@@ -96,3 +101,79 @@ def _exchange_code(self, code: str) -> str:
         token_data = response.json()
         logger.info("Authentication successful")
         return token_data["access_token"]
+
+
+class DirectGrantAuthenticator:
+    def __init__(self, session: requests.Session):
+        self.session = session
+
+    def login(self) -> str:
+        if not USERNAME or not PASSWORD:
+            raise Exception(
+                "USERNAME and PASSWORD environment variables are required for non-interactive authentication",
+            )
+
+        logger.info("Authenticating with username/password...")
+
+        token_url = (
+            f"{KEYCLOAK_URL}/realms/{REALM}/protocol/openid-connect/token"
+        )
+        data = {
+            "grant_type": "password",
+            "client_id": CLIENT_ID,
+            "username": USERNAME,
+            "password": PASSWORD,
+            "scope": "openid profile email organization",
+        }
+
+        if CLIENT_SECRET:
+            data["client_secret"] = CLIENT_SECRET
+
+        max_retries = 5
+        retry_delay = 2
+
+        for attempt in range(max_retries):
+            try:
+                response = self.session.post(token_url, data=data, timeout=10)
+
+                if response.status_code == 200:
+                    token_data = response.json()
+                    logger.info("Authentication successful")
+                    return token_data["access_token"]
+
+                if response.status_code == 503 or response.status_code == 502:
+                    if attempt < max_retries - 1:
+                        logger.warning(
+                            f"Keycloak not ready (HTTP {response.status_code}). "
+                            f"Retrying in {retry_delay} seconds... (attempt {attempt + 1}/{max_retries})",
+                        )
+                        time.sleep(retry_delay)
+                        continue
+
+                try:
+                    error_data = response.json()
+                    error_msg = error_data.get(
+                        "error_description",
+                        error_data.get("error", "Unknown error"),
+                    )
+                except Exception:
+                    error_msg = (
+                        f"HTTP {response.status_code}: {response.text[:100]}"
+                    )
+
+                raise Exception(
+                    f"Authentication failed: {error_msg}. "
+                    f"Make sure the client '{CLIENT_ID}' has 'Direct Access Grants' enabled in Keycloak "
+                    f"and that USERNAME/PASSWORD are correct.",
+                )
+            except requests.exceptions.RequestException as e:
+                if attempt < max_retries - 1:
+                    logger.warning(
+                        f"Connection error during authentication: {e}. "
+                        f"Retrying in {retry_delay} seconds... (attempt {attempt + 1}/{max_retries})",
+                    )
+                    time.sleep(retry_delay)
+                    continue
+                raise
+
+        raise Exception("Authentication failed after multiple retries")

simulator/config.py

@@ -8,7 +8,11 @@
 KEYCLOAK_URL = os.getenv("KEYCLOAK_URL", "http://localhost:8080")
 API_URL = os.getenv("API_URL", "http://localhost:8000/graphql")
 REALM = os.getenv("REALM", "tasks")
+USE_DIRECT_GRANT = os.getenv("USE_DIRECT_GRANT", "false").lower() == "true"
 CLIENT_ID = os.getenv("CLIENT_ID", "tasks-web")
+CLIENT_SECRET = os.getenv("CLIENT_SECRET", "")
+USERNAME = os.getenv("USERNAME", "test")
+PASSWORD = os.getenv("PASSWORD", "test")
 
 CALLBACK_PORT = 8999
 REDIRECT_URI = f"http://localhost:{CALLBACK_PORT}/callback"

simulator/graphql_client.py

@@ -1,14 +1,17 @@
 from typing import Optional
-from authentication import InteractiveAuthenticator
-from config import API_URL, logger
+from authentication import DirectGrantAuthenticator, InteractiveAuthenticator
+from config import API_URL, PASSWORD, USE_DIRECT_GRANT, USERNAME, logger
 import requests
 
 
 class GraphQLClient:
     def __init__(self):
         self.token: Optional[str] = None
         self.session = requests.Session()
-        self.authenticator = InteractiveAuthenticator(self.session)
+        if USE_DIRECT_GRANT and USERNAME and PASSWORD:
+            self.authenticator = DirectGrantAuthenticator(self.session)
+        else:
+            self.authenticator = InteractiveAuthenticator(self.session)
 
     def query(self, query: str, variables: dict = None) -> dict:
         if not self.token:

simulator/location_manager.py

@@ -1,20 +1,20 @@
-from typing import List, Optional
-from graphql_client import GraphQLClient
+import random
+
 from config import logger
 from data import RandomDataGenerator
-import random
+from graphql_client import GraphQLClient
 
 
 class LocationManager:
     def __init__(self, client: GraphQLClient):
         self.client = client
-        self.all_locations: List[dict] = []
-        self.locations: List[str] = []
-        self.beds: List[str] = []
-        self.rooms: List[str] = []
-        self.teams: List[str] = []
-        self.wards: List[str] = []
-        self.clinics: List[str] = []
+        self.all_locations: list[dict] = []
+        self.locations: list[str] = []
+        self.beds: list[str] = []
+        self.rooms: list[str] = []
+        self.teams: list[str] = []
+        self.wards: list[str] = []
+        self.clinics: list[str] = []
 
     def _log_errors(self, context: str, response: dict) -> None:
         if "errors" in response:
@@ -41,7 +41,9 @@ def load_locations(self) -> None:
         self.all_locations = data.get("locationNodes", [])
 
         self.locations = [
-            loc["id"] for loc in self.all_locations if loc["kind"] in ["BED", "ROOM"]
+            loc["id"]
+            for loc in self.all_locations
+            if loc["kind"] in ["BED", "ROOM"]
         ]
         self.beds = [
             loc["id"] for loc in self.all_locations if loc["kind"] == "BED"
@@ -61,11 +63,11 @@ def load_locations(self) -> None:
 
         logger.info(
             f"Locations loaded: {len(self.beds)} beds, {len(self.rooms)} rooms, "
-            f"{len(self.teams)} teams, {len(self.wards)} wards, {len(self.clinics)} clinics"
+            f"{len(self.teams)} teams, {len(self.wards)} wards, {len(self.clinics)} clinics",
         )
 
     def print_structure(self) -> None:
-        logger.info("\n" + "=" * 60)
+        logger.info("=" * 60)
         logger.info("LOCATION STRUCTURE")
         logger.info("=" * 60)
 
@@ -94,25 +96,34 @@ def print_tree(node_id: str, prefix: str = "", is_last: bool = True):
                 "ROOM": "🚪",
                 "BED": "🛏️",
             }.get(node["kind"], "📍")
-            print(f"{prefix}{connector}{kind_icon} {node['title']} ({node['kind']})")
+            print(
+                f"{prefix}{connector}{kind_icon} {node['title']} ({node['kind']})",
+            )
 
-            children = sorted(children_map[node_id], key=lambda cid: nodes[cid]["title"])
+            children = sorted(
+                children_map[node_id],
+                key=lambda cid: nodes[cid]["title"],
+            )
             for i, child_id in enumerate(children):
                 is_last_child = i == len(children) - 1
                 new_prefix = prefix + ("    " if is_last else "│   ")
                 print_tree(child_id, new_prefix, is_last_child)
 
-        for i, root_id in enumerate(sorted(roots, key=lambda rid: nodes[rid]["title"])):
+        for i, root_id in enumerate(
+            sorted(roots, key=lambda rid: nodes[rid]["title"]),
+        ):
             print_tree(root_id, "", i == len(roots) - 1)
         print("\n")
 
     def create_location(
         self,
         title: str,
         kind: str,
-        parent_id: Optional[str] = None,
-    ) -> Optional[str]:
-        logger.debug(f"Location creation requested: {title} ({kind}) - not implemented via API")
+        parent_id: str | None = None,
+    ) -> str | None:
+        logger.debug(
+            f"Location creation requested: {title} ({kind}) - not implemented via API",
+        )
         return None
 
     def create_rooms_and_beds(
@@ -140,25 +151,27 @@ def ensure_hospital_structure(self) -> None:
 
         logger.warning(
             "No locations found. Locations should be created via scaffold data "
-            "(see backend/scaffold.py). The simulator will work with existing locations only."
+            "(see backend/scaffold.py). The simulator will work with existing locations only.",
         )
         self.load_locations()
 
-    def add_team_to_ward(self, ward_id: Optional[str] = None) -> Optional[str]:
+    def add_team_to_ward(self, ward_id: str | None = None) -> str | None:
         if not ward_id:
             if not self.wards:
                 logger.warning("No wards available to add team to")
                 return None
             ward_id = random.choice(self.wards)
 
-        # Get ward info
-        ward = next((loc for loc in self.all_locations if loc["id"] == ward_id), None)
+        ward = next(
+            (loc for loc in self.all_locations if loc["id"] == ward_id),
+            None,
+        )
         if not ward:
             return None
 
-        # Generate team name
         available_teams = [
-            name for name in RandomDataGenerator.team_names
+            name
+            for name in RandomDataGenerator.team_names
             if not any(
                 loc["title"] == name and loc.get("parentId") == ward_id
                 for loc in self.all_locations
@@ -171,21 +184,21 @@ def add_team_to_ward(self, ward_id: Optional[str] = None) -> Optional[str]:
 
         logger.info(
             f"Would add new team '{team_name}' to ward '{ward['title']}' "
-            f"(location creation via API not available - use scaffold data)"
+            f"(location creation via API not available - use scaffold data)",
         )
         return None
 
-    def get_random_bed(self) -> Optional[str]:
+    def get_random_bed(self) -> str | None:
         if not self.beds:
             return None
         return random.choice(self.beds)
 
-    def get_random_room(self) -> Optional[str]:
+    def get_random_room(self) -> str | None:
         if not self.rooms:
             return None
         return random.choice(self.rooms)
 
-    def get_random_location(self) -> Optional[str]:
+    def get_random_location(self) -> str | None:
         if not self.locations:
             return None
         return random.choice(self.locations)

simulator/patient_manager.py

@@ -88,8 +88,10 @@ def ensure_diagnosis_property(self) -> None:
 
     def create_patient(self, admit_directly: bool = False) -> Tuple[Optional[str], Optional[str]]:
         if not self.location_manager.clinics:
-            logger.warning("No clinics available to assign patient")
-            return None, None
+            self.location_manager.load_locations()
+            if not self.location_manager.clinics:
+                logger.warning("No clinics available to assign patient")
+                return None, None
 
         first, last = RandomDataGenerator.get_name()
         diagnosis = TreatmentPlanner.get_random_diagnosis()