[uss_qualifier] Add auth adapter expectations to authorization verification

monitoring/monitorlib/inspection.py

@@ -1,6 +1,9 @@
 import importlib
 import inspect
 import pkgutil
+from typing import Any, Optional
+
+from implicitdict import ImplicitDict
 
 _modules_imported = set()
 
@@ -47,3 +50,77 @@ def fullname(class_type: type) -> str:
 
 def calling_function_name(levels: int = 0) -> str:
     return inspect.stack()[levels + 1].function
+
+
+class AttributeValuePair(ImplicitDict):
+    name: str
+    """The attribute that is expected to have a particular value.
+
+    Nested attributes are accepted (e.g., `"foo.bar"`)."""
+
+    equals_string_value: Optional[str]
+    """The attribute value is this string."""
+
+    equals_number_value: Optional[float]
+    """The attribute value is exactly this number.  Note that this may not be the desirable behavior when comparing float values."""
+
+
+def _has_attr(obj: Any, attr_name: str) -> bool:
+    if "." in attr_name:
+        levels = attr_name.split(".")
+        if not hasattr(obj, levels[0]):
+            return False
+        return _has_attr(getattr(obj, levels[0]), ".".join(levels[1:]))
+    else:
+        return hasattr(obj, attr_name)
+
+
+def _get_attr_value(obj: Any, attr_name: str) -> Any:
+    if "." in attr_name:
+        base, remaining = attr_name.split(".", 1)
+        return _get_attr_value(getattr(obj, base), remaining)
+    else:
+        return getattr(obj, attr_name)
+
+
+def evaluate_attributes(
+    obj: Any,
+    expectations: list[AttributeValuePair],
+) -> list[str]:
+    """Evaluates an object against a set of AttributeValuePair expectations.
+
+    Returns:
+        A list of string descriptions detailing any failed expectations. An empty list signifies success.
+    """
+    failures: list[str] = []
+    for pair in expectations:
+        attr_name = pair.name
+        if not _has_attr(obj, attr_name):
+            failures.append(
+                f"Required attribute '{attr_name}' is entirely absent from the object."
+            )
+            continue
+
+        actual_val = _get_attr_value(obj, attr_name)
+
+        if "equals_string_value" in pair and pair.equals_string_value is not None:
+            if not isinstance(actual_val, str):
+                failures.append(
+                    f"Attribute '{attr_name}' expected to be of type 'str', but observed type '{type(actual_val).__name__}'."
+                )
+            elif actual_val != pair.equals_string_value:
+                failures.append(
+                    f"Attribute '{attr_name}': Expected string value '{pair.equals_string_value}', but observed '{actual_val}'."
+                )
+
+        if "equals_number_value" in pair and pair.equals_number_value is not None:
+            if not isinstance(actual_val, (int, float)):
+                failures.append(
+                    f"Attribute '{attr_name}' expected to be numeric, but observed type '{type(actual_val).__name__}'."
+                )
+            elif actual_val != pair.equals_number_value:
+                failures.append(
+                    f"Attribute '{attr_name}': Expected numeric value {pair.equals_number_value}, but observed {actual_val}."
+                )
+
+    return failures

monitoring/uss_qualifier/configurations/dev/access_tokens.yaml

@@ -36,24 +36,38 @@ v1:
                       equals_string_value: dummy
                     - claim: sub
                       equals_string_value: uss_qualifier
+        auth_adapter_expectations:
+          resource_type: resources.communications.AuthAdapterExpectationsResource
+          specification:
+            expectations:
+              # See monitoring/monitorlib/auth.py for the set of concrete AuthAdapter types.
+              - adapter_type: DummyOAuth
+                attribute_values:
+                  # Attributes are for the concrete AuthAdapter instance available to the GetAccessTokens scenario.
+                  # See monitoring/monitorlib/auth.py for the definitions of concrete auth adapters and their attributes.
+                  - name: _oauth_token_endpoint
+                    equals_string_value: http://oauth.authority.localutm:8085/token
 
     action:
       test_suite:
         suite_definition:
           name: Access tokens validation test suite
           resources:
             auth_adapter: resources.communications.AuthAdapterResource
-            expectations: resources.communications.AccessTokensExpectationsResource
+            token_expectations: resources.communications.AccessTokensExpectationsResource?
+            adapter_expectations: resources.communications.AuthAdapterExpectationsResource?
           actions:
             - test_scenario:
                 scenario_type: scenarios.interuss.communications.GetAccessTokens
                 resources:
                   auth_adapter: auth_adapter
-                  expectations: expectations
+                  token_expectations: token_expectations
+                  adapter_expectations: adapter_expectations
                 on_failure: Continue
         resources:
           auth_adapter: utm_auth
-          expectations: access_token_expectations
+          token_expectations: access_token_expectations
+          adapter_expectations: auth_adapter_expectations
 
     execution:
       stop_fast: false

monitoring/uss_qualifier/requirements/interuss/communications/authorization.md

@@ -6,6 +6,14 @@ When a test designer requires certain behavior from an authorization source/serv
 
 ## Requirements
 
+### <tt>AuthType</tt>
+
+The authorization source must be of a particular type.
+
+### <tt>AuthAdapterAttribute</tt>
+
+A particular attribute of the AuthAdapter providing authorization must satisfy the criteria specified.
+
 ### <tt>GenerateAccessToken</tt>
 
 When provided with a valid and well-formed request to generate an access token, the provider of an authorization source must ensure that an access token is generated as requested.

monitoring/uss_qualifier/resources/communications/__init__.py

@@ -1,5 +1,10 @@
 from .access_token_expectations import (
     AccessTokensExpectationsResource as AccessTokensExpectationsResource,
 )
-from .auth_adapter import AuthAdapterResource as AuthAdapterResource
+from .auth_adapter import (
+    AuthAdapterExpectationsResource as AuthAdapterExpectationsResource,
+)
+from .auth_adapter import (
+    AuthAdapterResource as AuthAdapterResource,
+)
 from .client_identity import ClientIdentityResource as ClientIdentityResource

monitoring/uss_qualifier/resources/communications/auth_adapter.py

@@ -5,6 +5,7 @@
 
 from monitoring.monitorlib import infrastructure
 from monitoring.monitorlib.auth import make_auth_adapter
+from monitoring.monitorlib.inspection import AttributeValuePair
 from monitoring.uss_qualifier.configurations.configuration import ParticipantID
 from monitoring.uss_qualifier.resources.resource import MissingResourceError, Resource
 
@@ -78,3 +79,28 @@ def assert_scopes_available(
                     f"AuthAdapterResource provided to {consumer_name} is not declared (in its resource specification) as authorized to obtain scope `{scope}` which it requires to {reason}.  Update `scopes_authorized` to include `{scope}` to provide this authorization.  {len(self.scopes)} scopes currently declared as authorized for AuthAdapterResource: {', '.join(self.scopes)}",
                     "<unknown>",
                 )
+
+
+class AuthAdapterExpectations(ImplicitDict):
+    adapter_type: Optional[str]
+    """The auth adapter is expected to be an instance of the class named by this type (e.g., `ServiceAccountImpersonation`)."""
+
+    attribute_values: Optional[list[AttributeValuePair]]
+    """Particular attributes of the auth adapter satisfy these criteria."""
+
+
+class AuthAdapterExpectationsSpecification(ImplicitDict):
+    expectations: list[AuthAdapterExpectations]
+
+
+class AuthAdapterExpectationsResource(Resource[AuthAdapterExpectationsSpecification]):
+    spec: AuthAdapterExpectationsSpecification
+
+    def __init__(
+        self,
+        specification: AuthAdapterExpectationsSpecification,
+        resource_origin: str,
+        **dependencies,
+    ):
+        self.spec = specification
+        super().__init__(specification, resource_origin, **dependencies)

monitoring/uss_qualifier/scenarios/interuss/communications/get_access_tokens.md

@@ -8,12 +8,30 @@ This scenario obtains one or more access tokens using an auth adapter and, optio
 
 ### auth_adapter
 
-An [`AuthAdapterResources`](../../../resources/communications/auth_adapter.py) used to get the access tokens
+An [`AuthAdapterResource`](../../../resources/communications/auth_adapter.py) used to get the access tokens
 
-### expectations
+### token_expectations
 
 An [`AccessTokensExpectationsResource`](../../../resources/communications/access_token_expectations.py) describing what is expected of the access tokens acquired
 
+### adapter_expectations
+
+An [`AuthAdapterExpectationsResource`](../../../resources/communications/auth_adapter.py) describing what is expected of the auth adapter used to acquire access tokens
+
+## Validate auth adapter test case
+
+### Validate auth adapter characteristics test step
+
+In this step, the characteristics of the auth_adapter resource supplied to this scenario are evaluated according to adapter_expectations.
+
+#### ⚠️ Auth adapter type check
+
+If the auth_adapter resource contains an auth adapter that differs from the type specified in adapter_expectations, the provider of the authorization source fails to meet **[interuss.communications.authorization.AuthType](../../../requirements/interuss/communications/authorization.md)**.
+
+#### ⚠️ Auth adapter attribute check
+
+If the auth_adapter resource contains an auth adapter with an attribute that does not match a criterion specified in adapter_expectations, the provider of the authorization source fails to meet **[interuss.communications.authorization.AuthAdapterAttribute](../../../requirements/interuss/communications/authorization.md)**.
+
 ## Validate access tokens test case
 
 ### Get access token test step

monitoring/uss_qualifier/scenarios/interuss/communications/get_access_tokens.py

@@ -1,14 +1,18 @@
 from datetime import UTC, datetime
+from typing import Optional
 
 import jwt
 
 from monitoring.monitorlib.auth import AccessTokenError
 from monitoring.monitorlib.auth_validation import fix_key
+from monitoring.monitorlib.inspection import evaluate_attributes, fullname
+from monitoring.uss_qualifier.configurations.configuration import ParticipantID
 from monitoring.uss_qualifier.resources.communications.access_token_expectations import (
     AccessTokensExpectationsResource,
     ClaimValuePair,
 )
 from monitoring.uss_qualifier.resources.communications.auth_adapter import (
+    AuthAdapterExpectationsResource,
     AuthAdapterResource,
 )
 from monitoring.uss_qualifier.scenarios.scenario import TestScenario
@@ -21,23 +25,76 @@ class GetAccessTokens(TestScenario):
     def __init__(
         self,
         auth_adapter: AuthAdapterResource,
-        expectations: AccessTokensExpectationsResource,
+        token_expectations: Optional[AccessTokensExpectationsResource] = None,
+        adapter_expectations: Optional[AuthAdapterExpectationsResource] = None,
     ):
         super().__init__()
         self._auth_adapter = auth_adapter
-        self._expectations = expectations
+        self._token_expectations = token_expectations
+        self._adapter_expectations = adapter_expectations
 
     def run(self, context: ExecutionContext):
         self.begin_test_scenario(context)
-        self.begin_test_case("Validate access tokens")
 
         participants = (
             [self._auth_adapter.participant_id]
             if self._auth_adapter.participant_id
             else []
         )
 
-        for expect in self._expectations.spec.expectations:
+        self._validate_auth_adapter(participants)
+        self._validate_token_expectations(participants)
+
+        self.end_test_scenario()
+
+    def _validate_auth_adapter(self, participants: list[ParticipantID]):
+        if not self._adapter_expectations:
+            return
+
+        self.begin_test_case("Validate auth adapter")
+        self.begin_test_step("Validate auth adapter characteristics")
+
+        for expect in self._adapter_expectations.spec.expectations:
+            # --- Adapter Type Expectation ---
+            if "adapter_type" in expect and expect.adapter_type:
+                with self.check(
+                    "Auth adapter type", participants=participants
+                ) as check:
+                    actual_type = type(self._auth_adapter.adapter).__name__
+                    actual_fullname = fullname(type(self._auth_adapter.adapter))
+                    if (
+                        expect.adapter_type != actual_type
+                        and expect.adapter_type != actual_fullname
+                    ):
+                        check.record_failed(
+                            summary=f"Auth adapter is of type '{actual_type}' instead of expected '{expect.adapter_type}'",
+                            details=f"Expected auth adapter to be an instance of '{expect.adapter_type}', but found '{actual_fullname}' ({actual_type})",
+                        )
+
+            # --- Attribute Values Expectation ---
+            if "attribute_values" in expect and expect.attribute_values:
+                with self.check(
+                    "Auth adapter attribute", participants=participants
+                ) as check:
+                    failures = evaluate_attributes(
+                        self._auth_adapter.adapter, expect.attribute_values
+                    )
+                    if failures:
+                        check.record_failed(
+                            summary="One or more auth adapter attribute assertions failed",
+                            details="\n".join(failures),
+                        )
+
+        self.end_test_step()
+        self.end_test_case()
+
+    def _validate_token_expectations(self, participants: list[ParticipantID]):
+        if not self._token_expectations:
+            return
+
+        self.begin_test_case("Validate access tokens")
+
+        for expect in self._token_expectations.spec.expectations:
             # 1. Retrieve the access token and validate structural integrity as a JWT.
             self.begin_test_step("Get access token")
             token: str | None = None
@@ -86,7 +143,9 @@ def run(self, context: ExecutionContext):
                         "Token header value", participants=participants
                     ) as check:
                         failures = self._evaluate_claims(
-                            header, expect.expectations.has_header_values, request_time
+                            header,
+                            expect.expectations.has_header_values,
+                            request_time,
                         )
                         if failures:
                             check.record_failed(
@@ -141,7 +200,9 @@ def run(self, context: ExecutionContext):
                         "Token payload claim value", participants=participants
                     ) as check:
                         failures = self._evaluate_claims(
-                            payload, expect.expectations.has_claim_values, request_time
+                            payload,
+                            expect.expectations.has_claim_values,
+                            request_time,
                         )
                         if failures:
                             check.record_failed(
@@ -152,7 +213,6 @@ def run(self, context: ExecutionContext):
                 self.end_test_step()
 
         self.end_test_case()
-        self.end_test_scenario()
 
     def _evaluate_claims(
         self,

schemas/monitoring/monitorlib/inspection/AttributeValuePair.json

@@ -0,0 +1,33 @@
+{
+  "$id": "https://github.com/interuss/monitoring/blob/main/schemas/monitoring/monitorlib/inspection/AttributeValuePair.json",
+  "$schema": "https://json-schema.org/draft/2020-12/schema",
+  "description": "monitoring.monitorlib.inspection.AttributeValuePair, as defined in monitoring/monitorlib/inspection.py",
+  "properties": {
+    "$ref": {
+      "description": "Path to content that replaces the $ref",
+      "type": "string"
+    },
+    "equals_number_value": {
+      "description": "The attribute value is exactly this number.  Note that this may not be the desirable behavior when comparing float values.",
+      "type": [
+        "number",
+        "null"
+      ]
+    },
+    "equals_string_value": {
+      "description": "The attribute value is this string.",
+      "type": [
+        "string",
+        "null"
+      ]
+    },
+    "name": {
+      "description": "The attribute that is expected to have a particular value.\n\nNested attributes are accepted (e.g., `\"foo.bar\"`).",
+      "type": "string"
+    }
+  },
+  "required": [
+    "name"
+  ],
+  "type": "object"
+}