feat(stratus): add credential reference element on contracts (#7850)

stratus/stratus/contracts/__init__.py

@@ -32,6 +32,7 @@
     SupportedLanguage,
     prepare_contracts,
 )
+from pyoaev.credential.utils import build_single_referenced_credential_element
 from pyoaev.security_domain.types import SecurityDomains
 
 from stratus.contracts.platforms import (  # noqa: F401 (re-exported)
@@ -140,6 +141,9 @@ def _make_contract(
     elements: List[ContractElement] = [
         _credential_element(cred) for cred in platform.cred_fields
     ]
+    # Centralized credential-reference field: lets the OpenAEV inject form ask
+    # for a stored credential, filtered on the platform credential type.
+    elements.append(build_single_referenced_credential_element(platform.key))
     elements.extend(extra_fields)
     elements.append(_expectations_element())
 

stratus/test/test_contracts.py

@@ -1,6 +1,9 @@
 import json
 from unittest import TestCase
 
+from pyoaev.contracts.contract_config import ContractFieldKey, ContractFieldType
+from pyoaev.credential.types import CredentialType
+
 from stratus.contracts import (
     CONTRACT_REGISTRY,
     CONTRACT_TYPE,
@@ -11,6 +14,19 @@
 from stratus.contracts.platforms import PLATFORMS, PLATFORMS_BY_KEY
 from stratus.contracts.techniques import TECHNIQUES
 
+CREDENTIAL_REFERENCE_KEY = ContractFieldKey.CredentialReference.value
+
+# entra-id and k8s have no dedicated platform credential type: a None value is
+# intentional, the frontend then applies no credential filter.
+EXPECTED_CREDENTIAL_TYPE_BY_PLATFORM = {
+    "aws": CredentialType.CLOUD_AWS.value,
+    "eks": CredentialType.CLOUD_AWS.value,
+    "azure": CredentialType.CLOUD_AZURE.value,
+    "entra-id": None,
+    "gcp": CredentialType.CLOUD_GCP.value,
+    "k8s": None,
+}
+
 
 class ContractsTest(TestCase):
     def setUp(self):
@@ -75,3 +91,68 @@ def test_labels_are_platform_prefixed(self):
             self.assertTrue(
                 contract["contract_labels"]["en"].startswith(platform.label + " - ")
             )
+
+    def _credential_reference_field(self, content):
+        fields = [f for f in content["fields"] if f["key"] == CREDENTIAL_REFERENCE_KEY]
+        self.assertEqual(len(fields), 1)
+        return fields[0]
+
+    def test_every_contract_exposes_exactly_one_credential_reference(self):
+        for content in self.content_by_id.values():
+            field = self._credential_reference_field(content)
+            self.assertEqual(field["type"], ContractFieldType.CredentialReference.value)
+
+    def test_credential_reference_is_mandatory_and_single_valued(self):
+        for content in self.content_by_id.values():
+            field = self._credential_reference_field(content)
+            self.assertFalse(field["mandatory"])
+            self.assertFalse(field["multiple"])
+
+    def test_custom_contracts_credential_reference_type_matches_platform(self):
+        for platform in PLATFORMS:
+            content = self.content_by_id[platform.custom_contract_id]
+            field = self._credential_reference_field(content)
+            self.assertEqual(
+                field["credential_reference_type"],
+                EXPECTED_CREDENTIAL_TYPE_BY_PLATFORM[platform.key],
+            )
+
+    def test_technique_contracts_credential_reference_type_matches_platform(self):
+        for technique in TECHNIQUES:
+            content = self.content_by_id[technique_contract_id(technique.id)]
+            field = self._credential_reference_field(content)
+            self.assertEqual(
+                field["credential_reference_type"],
+                EXPECTED_CREDENTIAL_TYPE_BY_PLATFORM[technique.platform],
+            )
+
+    def test_every_platform_has_an_expected_credential_type(self):
+        self.assertEqual(
+            {p.key for p in PLATFORMS}, set(EXPECTED_CREDENTIAL_TYPE_BY_PLATFORM)
+        )
+
+    def test_credential_reference_does_not_collide_with_platform_cred_fields(self):
+        for platform in PLATFORMS:
+            cred_keys = {cred.key for cred in platform.cred_fields}
+            self.assertNotIn(CREDENTIAL_REFERENCE_KEY, cred_keys)
+
+    def test_credential_reference_sits_after_credentials_before_expectations(self):
+        for technique in TECHNIQUES:
+            platform = PLATFORMS_BY_KEY[technique.platform]
+            content = self.content_by_id[technique_contract_id(technique.id)]
+            keys = [f["key"] for f in content["fields"]]
+            self.assertEqual(
+                keys.index(CREDENTIAL_REFERENCE_KEY), len(platform.cred_fields)
+            )
+            self.assertLess(
+                keys.index(CREDENTIAL_REFERENCE_KEY), keys.index("expectations")
+            )
+
+    def test_custom_contracts_keep_credential_reference_before_technique_field(self):
+        for platform in PLATFORMS:
+            content = self.content_by_id[platform.custom_contract_id]
+            keys = [f["key"] for f in content["fields"]]
+            self.assertLess(
+                keys.index(CREDENTIAL_REFERENCE_KEY),
+                keys.index(CUSTOM_TECHNIQUE_FIELD_KEY),
+            )