Address SonarCloud maintainability findings

timbl-ont · claude · timbl-ont · commit bacd860d7dec · 2026-07-06T22:10:37.000-04:00
- oauth_validator: use a dict literal instead of the dict() constructor
  (python:S7498).
- Extract the OAuth scope check from tenant_authentication into
  _enforce_tenant_scope to reduce cognitive complexity (python:S3776).
- Extract OAuth validation and settings mapping from AdminGroup.get_settings
  into _validate_admin_auth_args and _oauth_settings helpers, reducing its
  cognitive complexity and collapsing the nested if (S3776, S1066).
- Redirect demo-script ERROR messages to stderr (shell S7677).

Co-Authored-By: Claude Fable 5 &lt;noreply@anthropic.com&gt;
Signed-off-by: Tim Bloomfield &lt;tim.bloomfield@ontario.ca&gt;
diff --git a/acapy_agent/admin/decorators/auth.py b/acapy_agent/admin/decorators/auth.py
@@ -74,23 +74,9 @@ async def tenant_auth(request):
             return await handler(request)
 
         # OAuth path: token was validated by setup_context middleware.
-        # acapy:tenant or acapy:admin both grant tenant-level access.
-        # acapy:tenant:read grants read-only access (safe HTTP methods only).
         if has_auth_scopes(context):
-            scopes = get_auth_scopes(context)
-            if scopes & {"acapy:tenant", "acapy:admin"}:
-                return await handler(request)
-            if "acapy:tenant:read" in scopes:
-                if request.method in ("GET", "HEAD"):
-                    return await handler(request)
-                raise web.HTTPForbidden(
-                    reason="acapy:tenant:read scope does not permit write operations",
-                    text="acapy:tenant:read scope does not permit write operations",
-                )
-            raise web.HTTPForbidden(
-                reason="acapy:tenant scope required",
-                text="acapy:tenant scope required",
-            )
+            _enforce_tenant_scope(get_auth_scopes(context), request.method)
+            return await handler(request)
 
         authorization_header = request.headers.get("Authorization")
         header_admin_api_key = request.headers.get("x-api-key")
@@ -162,6 +148,27 @@ async def scope_check(request):
     return decorator
 
 
+def _enforce_tenant_scope(scopes: set, method: str) -> None:
+    """Authorize a tenant request by its OAuth scopes, else raise HTTPForbidden.
+
+    ``acapy:tenant`` or ``acapy:admin`` grant tenant-level access;
+    ``acapy:tenant:read`` grants read-only access (safe HTTP methods only).
+    """
+    if scopes & {"acapy:tenant", "acapy:admin"}:
+        return
+    if "acapy:tenant:read" in scopes:
+        if method in ("GET", "HEAD"):
+            return
+        raise web.HTTPForbidden(
+            reason="acapy:tenant:read scope does not permit write operations",
+            text="acapy:tenant:read scope does not permit write operations",
+        )
+    raise web.HTTPForbidden(
+        reason="acapy:tenant scope required",
+        text="acapy:tenant scope required",
+    )
+
+
 def _base_wallet_route_access(additional_routes: List[str], request_path: str) -> bool:
     """Check if request path matches additional routes."""
     additional_routes_pattern = (
diff --git a/acapy_agent/admin/oauth_validator.py b/acapy_agent/admin/oauth_validator.py
@@ -121,15 +121,15 @@ async def _validate_jwt(self, token: str) -> dict:
             None, self._jwks_client.get_signing_key_from_jwt, token
         )
 
-        decode_kwargs = dict(
-            algorithms=_SUPPORTED_ALGORITHMS,
+        decode_kwargs = {
+            "algorithms": _SUPPORTED_ALGORITHMS,
             # Without verify_aud=False, PyJWT rejects any token carrying an
             # aud claim when no expected audience is configured.
-            options={
+            "options": {
                 "require": ["exp", "iss", "sub"],
                 "verify_aud": self.audience is not None,
             },
-        )
+        }
         if self.issuer:
             decode_kwargs["issuer"] = self.issuer
         if self.audience:
diff --git a/acapy_agent/config/argparse.py b/acapy_agent/config/argparse.py
@@ -379,6 +379,76 @@ def add_arguments(self, parser: ArgumentParser):
             ),
         )
 
+    @staticmethod
+    def _validate_admin_auth_args(
+        args: Namespace, oauth_mode: bool, admin_api_key, admin_insecure_mode
+    ):
+        """Validate admin authentication argument combinations.
+
+        Raises:
+            ArgsParseError: if the admin auth flags are inconsistent.
+
+        """
+        if not oauth_mode:
+            if (admin_api_key and admin_insecure_mode) or not (
+                admin_api_key or admin_insecure_mode
+            ):
+                raise ArgsParseError(
+                    "Either --admin-api-key or --admin-insecure-mode "
+                    "must be set but not both, unless --oauth-enabled (or "
+                    "--oauth-jwks-uri / --oauth-introspection-endpoint) "
+                    "is configured."
+                )
+            return
+
+        if not (
+            getattr(args, "oauth_jwks_uri", None)
+            or getattr(args, "oauth_introspection_endpoint", None)
+        ):
+            raise ArgsParseError(
+                "OAuth mode requires a token validation method: set "
+                "--oauth-jwks-uri and/or --oauth-introspection-endpoint."
+            )
+        if getattr(args, "oauth_introspection_endpoint", None) and not getattr(
+            args, "oauth_introspection_client_id", None
+        ):
+            raise ArgsParseError(
+                "--oauth-introspection-endpoint requires --oauth-introspection-client-id."
+            )
+        if getattr(args, "oauth_jwks_uri", None) and not getattr(
+            args, "oauth_audience", None
+        ):
+            # Without an expected audience, any signature-valid token from the
+            # JWKS is accepted regardless of its intended recipient, allowing
+            # token reuse / confused-deputy on a shared AS.
+            raise ArgsParseError(
+                "--oauth-jwks-uri requires --oauth-audience so that JWT "
+                "access tokens are bound to this resource server (the "
+                "'aud' claim is verified)."
+            )
+
+    @staticmethod
+    def _oauth_settings(args: Namespace, oauth_mode: bool) -> dict:
+        """Build the oauth.* settings map from parsed arguments."""
+        settings = {}
+        if oauth_mode:
+            settings["admin.oauth_enabled"] = True
+            settings["oauth.http_timeout"] = getattr(args, "oauth_http_timeout", None)
+
+        arg_to_setting = {
+            "oauth_jwks_uri": "oauth.jwks_uri",
+            "oauth_issuer": "oauth.issuer",
+            "oauth_audience": "oauth.audience",
+            "oauth_introspection_endpoint": "oauth.introspection_endpoint",
+            "oauth_introspection_client_id": "oauth.introspection_client_id",
+            "oauth_introspection_client_secret": "oauth.introspection_client_secret",
+        }
+        for arg_name, setting_key in arg_to_setting.items():
+            value = getattr(args, arg_name, None)
+            if value:
+                settings[setting_key] = value
+        return settings
+
     def get_settings(self, args: Namespace):
         """Extract admin settings."""
         settings = {}
@@ -391,69 +461,13 @@ def get_settings(self, args: Namespace):
                 or getattr(args, "oauth_introspection_endpoint", None)
             )
 
-            if not oauth_mode:
-                if (admin_api_key and admin_insecure_mode) or not (
-                    admin_api_key or admin_insecure_mode
-                ):
-                    raise ArgsParseError(
-                        "Either --admin-api-key or --admin-insecure-mode "
-                        "must be set but not both, unless --oauth-enabled (or "
-                        "--oauth-jwks-uri / --oauth-introspection-endpoint) "
-                        "is configured."
-                    )
-            else:
-                if not (
-                    getattr(args, "oauth_jwks_uri", None)
-                    or getattr(args, "oauth_introspection_endpoint", None)
-                ):
-                    raise ArgsParseError(
-                        "OAuth mode requires a token validation method: set "
-                        "--oauth-jwks-uri and/or --oauth-introspection-endpoint."
-                    )
-                if getattr(args, "oauth_introspection_endpoint", None) and not getattr(
-                    args, "oauth_introspection_client_id", None
-                ):
-                    raise ArgsParseError(
-                        "--oauth-introspection-endpoint requires "
-                        "--oauth-introspection-client-id."
-                    )
-                if getattr(args, "oauth_jwks_uri", None) and not getattr(
-                    args, "oauth_audience", None
-                ):
-                    # Without an expected audience, any signature-valid token from
-                    # the JWKS is accepted regardless of its intended recipient,
-                    # allowing token reuse / confused-deputy on a shared AS.
-                    raise ArgsParseError(
-                        "--oauth-jwks-uri requires --oauth-audience so that JWT "
-                        "access tokens are bound to this resource server (the "
-                        "'aud' claim is verified)."
-                    )
+            self._validate_admin_auth_args(
+                args, oauth_mode, admin_api_key, admin_insecure_mode
+            )
 
             settings["admin.admin_api_key"] = admin_api_key
             settings["admin.admin_insecure_mode"] = admin_insecure_mode
-
-            if oauth_mode:
-                settings["admin.oauth_enabled"] = True
-                settings["oauth.http_timeout"] = getattr(args, "oauth_http_timeout", None)
-
-            if getattr(args, "oauth_jwks_uri", None):
-                settings["oauth.jwks_uri"] = args.oauth_jwks_uri
-            if getattr(args, "oauth_issuer", None):
-                settings["oauth.issuer"] = args.oauth_issuer
-            if getattr(args, "oauth_audience", None):
-                settings["oauth.audience"] = args.oauth_audience
-            if getattr(args, "oauth_introspection_endpoint", None):
-                settings["oauth.introspection_endpoint"] = (
-                    args.oauth_introspection_endpoint
-                )
-            if getattr(args, "oauth_introspection_client_id", None):
-                settings["oauth.introspection_client_id"] = (
-                    args.oauth_introspection_client_id
-                )
-            if getattr(args, "oauth_introspection_client_secret", None):
-                settings["oauth.introspection_client_secret"] = (
-                    args.oauth_introspection_client_secret
-                )
+            settings.update(self._oauth_settings(args, oauth_mode))
 
             settings["admin.enabled"] = True
             settings["admin.host"] = args.admin[0]
diff --git a/demo/demo-authserver/scripts/get-user-token.sh b/demo/demo-authserver/scripts/get-user-token.sh
@@ -79,7 +79,7 @@ WALLET_ID=$(curl -sf \
   | jq -r '.[] | select(.name == "wallet-id") | .config["claim.value"] // empty')
 
 if [[ -z "${WALLET_ID}" || "${WALLET_ID}" == "PLACEHOLDER_WALLET_ID" ]]; then
-  echo "ERROR: wallet_id not set on '${TENANT_CLIENT_ID}'. Run ./scripts/setup-tenant.sh first."
+  echo "ERROR: wallet_id not set on '${TENANT_CLIENT_ID}'. Run ./scripts/setup-tenant.sh first." >&2
   exit 1
 fi
 
@@ -225,7 +225,7 @@ print(result[0] if result else '')
 ")
 
 if [[ -z "${AUTH_CODE}" ]]; then
-  echo "ERROR: No authorization code received."
+  echo "ERROR: No authorization code received." >&2
   exit 1
 fi
 
@@ -244,8 +244,8 @@ TOKEN_RESPONSE=$(curl -s -X POST "${TOKEN_ENDPOINT}" \
 ACCESS_TOKEN=$(echo "${TOKEN_RESPONSE}" | jq -r '.access_token // empty')
 
 if [[ -z "${ACCESS_TOKEN}" ]]; then
-  echo "ERROR: Token exchange failed."
-  echo "       Response: ${TOKEN_RESPONSE}"
+  echo "ERROR: Token exchange failed." >&2
+  echo "       Response: ${TOKEN_RESPONSE}" >&2
   exit 1
 fi
 
diff --git a/demo/demo-authserver/scripts/setup-tenant.sh b/demo/demo-authserver/scripts/setup-tenant.sh
@@ -97,7 +97,7 @@ if [[ -z "${WALLET_ID}" ]]; then
 fi
 
 if [[ -z "${WALLET_ID}" ]]; then
-  echo "ERROR: Could not create or find wallet '${WALLET_NAME}'."
+  echo "ERROR: Could not create or find wallet '${WALLET_NAME}'." >&2
   exit 1
 fi
 echo "    wallet_id: ${WALLET_ID}"
@@ -112,7 +112,7 @@ KC_CLIENT_UUID=$(curl -sf \
   | jq -r '.[0].id // empty')
 
 if [[ -z "${KC_CLIENT_UUID}" ]]; then
-  echo "ERROR: Client '${TENANT_CLIENT_ID}' not found in Keycloak realm '${KEYCLOAK_REALM}'."
+  echo "ERROR: Client '${TENANT_CLIENT_ID}' not found in Keycloak realm '${KEYCLOAK_REALM}'." >&2
   exit 1
 fi
 echo "    client UUID: ${KC_CLIENT_UUID}"
@@ -152,7 +152,7 @@ CREATE_STATUS=$(curl -s -o /dev/null -w "%{http_code}" -X POST \
     }
   }")
 if [[ "${CREATE_STATUS}" != "201" ]]; then
-  echo "ERROR: Failed to create wallet-id mapper (HTTP ${CREATE_STATUS})."
+  echo "ERROR: Failed to create wallet-id mapper (HTTP ${CREATE_STATUS})." >&2
   exit 1
 fi
 echo "    Mapper created (HTTP ${CREATE_STATUS})."
@@ -164,7 +164,7 @@ ACTUAL_WALLET_ID=$(curl -sf \
   | jq -r '.[] | select(.name == "wallet-id") | .config["claim.value"] // empty')
 echo "    Verified claim.value in Keycloak: ${ACTUAL_WALLET_ID}"
 if [[ "${ACTUAL_WALLET_ID}" != "${WALLET_ID}" ]]; then
-  echo "ERROR: Keycloak mapper value does not match wallet_id (${ACTUAL_WALLET_ID} != ${WALLET_ID})."
+  echo "ERROR: Keycloak mapper value does not match wallet_id (${ACTUAL_WALLET_ID} != ${WALLET_ID})." >&2
   exit 1
 fi
 
@@ -268,8 +268,8 @@ WALLET_CREATE_SCOPE_UUID=$(curl -sf \
   | jq -r '.[] | select(.name == "acapy:wallet:create") | .id // empty')
 
 if [[ -z "${WALLET_CREATE_SCOPE_UUID}" ]]; then
-  echo "ERROR: 'acapy:wallet:create' client scope not found in realm '${KEYCLOAK_REALM}'."
-  echo "       Ensure the realm was imported from keycloak/realm-export.json."
+  echo "ERROR: 'acapy:wallet:create' client scope not found in realm '${KEYCLOAK_REALM}'." >&2
+  echo "       Ensure the realm was imported from keycloak/realm-export.json." >&2
   exit 1
 fi
 echo "    acapy:wallet:create scope UUID: ${WALLET_CREATE_SCOPE_UUID}"
@@ -281,7 +281,7 @@ ASSIGN_STATUS=$(curl -s -o /dev/null -w "%{http_code}" -X PUT \
 if [[ "${ASSIGN_STATUS}" == "204" || "${ASSIGN_STATUS}" == "409" ]]; then
   echo "    acapy:wallet:create scope assigned (HTTP ${ASSIGN_STATUS})."
 else
-  echo "ERROR: Failed to assign acapy:wallet:create scope (HTTP ${ASSIGN_STATUS})."
+  echo "ERROR: Failed to assign acapy:wallet:create scope (HTTP ${ASSIGN_STATUS})." >&2
   exit 1
 fi
 
diff --git a/demo/demo-authserver/scripts/test-oauth-scopes.sh b/demo/demo-authserver/scripts/test-oauth-scopes.sh
@@ -153,8 +153,8 @@ WALLET_ID=$(echo "${TENANT_TOKEN}" \
 
 if [[ -z "${WALLET_ID}" || "${WALLET_ID}" == "PLACEHOLDER_WALLET_ID" ]]; then
   echo ""
-  echo "ERROR: tenant token does not contain a valid wallet_id claim."
-  echo "       Run ./scripts/setup-tenant.sh first."
+  echo "ERROR: tenant token does not contain a valid wallet_id claim." >&2
+  echo "       Run ./scripts/setup-tenant.sh first." >&2
   exit 1
 fi
 echo "    wallet_id:    ${WALLET_ID}"
