Add required product association for APIs

Author: simonkurtz-MSFT

samples/authX-pro/create.ipynb

@@ -148,15 +148,15 @@
     "hremployees_api_path = f'/{api_prefix}employees'\n",
     "hremployees_get = GET_APIOperation('Gets the employees', utils.read_policy_xml('./hr_get.xml'))\n",
     "hremployees_post = POST_APIOperation('Creates a new employee', utils.read_policy_xml('./hr_post.xml'))\n",
-    "hremployees = API(f'{api_prefix}Employees', 'Employees Pro', hremployees_api_path, 'This is a Human Resources API for employee information', utils.read_policy_xml(DEFAULT_XML_POLICY_PATH), \n",
-    "                  operations = [hremployees_get, hremployees_post], tags = tags, productNames = [hr_product_name], subscriptionRequired = True)\n",
+    "hremployees = API(f'{api_prefix}Employees', 'Employees Pro', hremployees_api_path, 'This is a Human Resources API for employee information', utils.read_policy_xml(REQUIRE_PRODUCT_XML_POLICY_PATH), \n",
+    "                  operations = [hremployees_get, hremployees_post], tags = tags, productNames = [hr_product_name], subscriptionRequired = False)\n",
     "\n",
     "# Benefits (HR)\n",
     "hrbenefits_api_path = f'/{api_prefix}benefits'\n",
     "hrbenefits_get = GET_APIOperation('Gets employee benefits', utils.read_policy_xml('./hr_get.xml'))\n",
     "hrbenefits_post = POST_APIOperation('Creates employee benefits', utils.read_policy_xml('./hr_post.xml'))\n",
-    "hrbenefits = API(f'{api_prefix}Benefits', 'Benefits Pro', hrbenefits_api_path, 'This is a Human Resources API for employee benefits', utils.read_policy_xml(DEFAULT_XML_POLICY_PATH), \n",
-    "                 operations = [hrbenefits_get, hrbenefits_post], tags = tags, productNames = [hr_product_name], subscriptionRequired = True)\n",
+    "hrbenefits = API(f'{api_prefix}Benefits', 'Benefits Pro', hrbenefits_api_path, 'This is a Human Resources API for employee benefits', utils.read_policy_xml(REQUIRE_PRODUCT_XML_POLICY_PATH), \n",
+    "                 operations = [hrbenefits_get, hrbenefits_post], tags = tags, productNames = [hr_product_name], subscriptionRequired = False)\n",
     "\n",
     "# APIs Array\n",
     "apis: List[API] = [hremployees, hrbenefits]\n",

shared/apim-policies/require-product.xml

@@ -0,0 +1,28 @@
+<!--
+    Policy that ensures the API is called with a product subscription.
+    This policy validates that the request includes a valid product context.
+    Returns 403 Unauthorized with JSON error response if no product subscription is found.
+-->
+<policies>
+    <inbound>
+        <base />
+        <!-- Validate that a product subscription is present -->
+        <choose>
+            <when condition="@(context.Product == null)">
+                <return-response>
+                    <set-status code="403" reason="Unauthorized" />
+                    <set-body>"This API requires a valid product subscription to access."</set-body>
+                </return-response>
+            </when>
+        </choose>
+    </inbound>
+    <backend>
+        <base />
+    </backend>
+    <outbound>
+        <base />
+    </outbound>
+    <on-error>
+        <base />
+    </on-error>
+</policies>

shared/bicep/modules/apim/v1/api.bicep

@@ -60,12 +60,12 @@ resource apimApi 'Microsoft.ApiManagement/service/apis@2024-06-01-preview' = {
     path: api.path
     protocols: [
       'https'
-    ]
+    ]   
     subscriptionKeyParameterNames: {
       header: 'api-key'
       query: 'api-key'
     }
-    subscriptionRequired: false
+    subscriptionRequired: api.?subscriptionRequired ?? false
     type: 'http'
   }
 }

shared/python/apimtypes.py

@@ -14,6 +14,7 @@
 # These paths are relative to the infrastructure and samples
 SHARED_XML_POLICY_BASE_PATH         = '../../shared/apim-policies'
 DEFAULT_XML_POLICY_PATH             = f'{SHARED_XML_POLICY_BASE_PATH}/default.xml'
+REQUIRE_PRODUCT_XML_POLICY_PATH     = f'{SHARED_XML_POLICY_BASE_PATH}/require-product.xml'
 HELLO_WORLD_XML_POLICY_PATH         = f'{SHARED_XML_POLICY_BASE_PATH}/hello-world.xml'
 REQUEST_HEADERS_XML_POLICY_PATH     = f'{SHARED_XML_POLICY_BASE_PATH}/request-headers.xml'
 BACKEND_XML_POLICY_PATH             = f'{SHARED_XML_POLICY_BASE_PATH}/backend.xml'
@@ -121,11 +122,14 @@ class API:
     operations: Optional[List['APIOperation']] = None
     tags: Optional[List[str]] = None
     productNames: Optional[List[str]] = None
+    subscriptionRequired: bool = False
+
     # ------------------------------
     #    CONSTRUCTOR
     # ------------------------------
 
-    def __init__(self, name: str, displayName: str, path: str, description: str, policyXml: Optional[str] = None, operations: Optional[List['APIOperation']] = None, tags: Optional[List[str]] = None, productNames: Optional[List[str]] = None):
+    def __init__(self, name: str, displayName: str, path: str, description: str, policyXml: Optional[str] = None, operations: Optional[List['APIOperation']] = None, tags: Optional[List[str]] = None, 
+                 productNames: Optional[List[str]] = None, subscriptionRequired: bool = False):
         self.name = name
         self.displayName = displayName
         self.path = path
@@ -134,6 +138,8 @@ def __init__(self, name: str, displayName: str, path: str, description: str, pol
         self.operations = operations if operations is not None else []
         self.tags = tags if tags is not None else []
         self.productNames = productNames if productNames is not None else []
+        self.subscriptionRequired = subscriptionRequired
+
     # ------------------------------
     #    PUBLIC METHODS
     # ------------------------------
@@ -144,7 +150,8 @@ def to_dict(self) -> dict:
             "displayName": self.displayName,
             "path": self.path,
             "description": self.description,
-            "operations": [op.to_dict() for op in self.operations] if self.operations else []
+            "operations": [op.to_dict() for op in self.operations] if self.operations else [],
+            "subscriptionRequired": self.subscriptionRequired
         }
 
         if self.policyXml is not None:

tests/python/test_apimtypes.py

@@ -531,3 +531,153 @@ def test_product_repr():
     assert "Product" in result
     assert "hr" in result
     assert "Human Resources" in result
+
+@pytest.mark.unit
+def test_api_subscription_required_default():
+    """Test that API object has subscriptionRequired defaulting to False."""
+    api = apimtypes.API(
+        name = EXAMPLE_NAME,
+        displayName = EXAMPLE_DISPLAY_NAME,
+        path = EXAMPLE_PATH,
+        description = EXAMPLE_DESCRIPTION,
+        policyXml = EXAMPLE_POLICY_XML,
+        operations = None
+    )
+    assert api.subscriptionRequired == False
+
+@pytest.mark.unit
+def test_api_subscription_required_explicit_false():
+    """Test creation of API object with explicit subscriptionRequired=False."""
+    api = apimtypes.API(
+        name = EXAMPLE_NAME,
+        displayName = EXAMPLE_DISPLAY_NAME,
+        path = EXAMPLE_PATH,
+        description = EXAMPLE_DESCRIPTION,
+        policyXml = EXAMPLE_POLICY_XML,
+        operations = None,
+        subscriptionRequired = False
+    )
+    assert api.subscriptionRequired == False
+
+@pytest.mark.unit
+def test_api_subscription_required_explicit_true():
+    """Test creation of API object with explicit subscriptionRequired=True."""
+    api = apimtypes.API(
+        name = EXAMPLE_NAME,
+        displayName = EXAMPLE_DISPLAY_NAME,
+        path = EXAMPLE_PATH,
+        description = EXAMPLE_DESCRIPTION,
+        policyXml = EXAMPLE_POLICY_XML,
+        operations = None,
+        subscriptionRequired = True
+    )
+    assert api.subscriptionRequired == True
+
+@pytest.mark.unit
+def test_api_to_dict_includes_subscription_required_when_true():
+    """Test that to_dict includes subscriptionRequired when True."""
+    api = apimtypes.API(
+        name = EXAMPLE_NAME,
+        displayName = EXAMPLE_DISPLAY_NAME,
+        path = EXAMPLE_PATH,
+        description = EXAMPLE_DESCRIPTION,
+        policyXml = EXAMPLE_POLICY_XML,
+        operations = None,
+        subscriptionRequired = True
+    )
+    d = api.to_dict()
+    assert "subscriptionRequired" in d
+    assert d["subscriptionRequired"] == True
+
+@pytest.mark.unit
+def test_api_to_dict_includes_subscription_required_when_false():
+    """Test that to_dict includes subscriptionRequired when explicitly False."""
+    api = apimtypes.API(
+        name = EXAMPLE_NAME,
+        displayName = EXAMPLE_DISPLAY_NAME,
+        path = EXAMPLE_PATH,
+        description = EXAMPLE_DESCRIPTION,
+        policyXml = EXAMPLE_POLICY_XML,
+        operations = None,
+        subscriptionRequired = False
+    )
+    d = api.to_dict()
+    assert "subscriptionRequired" in d
+    assert d["subscriptionRequired"] == False
+
+@pytest.mark.unit
+def test_api_equality_with_subscription_required():
+    """Test equality comparison for API objects with different subscriptionRequired values."""
+    api1 = apimtypes.API(
+        name = EXAMPLE_NAME,
+        displayName = EXAMPLE_DISPLAY_NAME,
+        path = EXAMPLE_PATH,
+        description = EXAMPLE_DESCRIPTION,
+        policyXml = EXAMPLE_POLICY_XML,
+        operations = None,
+        subscriptionRequired = True
+    )
+    api2 = apimtypes.API(
+        name = EXAMPLE_NAME,
+        displayName = EXAMPLE_DISPLAY_NAME,
+        path = EXAMPLE_PATH,
+        description = EXAMPLE_DESCRIPTION,
+        policyXml = EXAMPLE_POLICY_XML,
+        operations = None,
+        subscriptionRequired = True
+    )
+    api3 = apimtypes.API(
+        name = EXAMPLE_NAME,
+        displayName = EXAMPLE_DISPLAY_NAME,
+        path = EXAMPLE_PATH,
+        description = EXAMPLE_DESCRIPTION,
+        policyXml = EXAMPLE_POLICY_XML,
+        operations = None,
+        subscriptionRequired = False
+    )
+    
+    # Same subscriptionRequired values should be equal
+    assert api1 == api2
+    
+    # Different subscriptionRequired values should not be equal
+    assert api1 != api3
+
+@pytest.mark.unit
+def test_api_with_all_properties():
+    """Test creation of API object with all properties including subscriptionRequired."""
+    tags = ["tag1", "tag2"]
+    product_names = ["product1", "product2"]
+    api = apimtypes.API(
+        name = EXAMPLE_NAME,
+        displayName = EXAMPLE_DISPLAY_NAME,
+        path = EXAMPLE_PATH,
+        description = EXAMPLE_DESCRIPTION,
+        policyXml = EXAMPLE_POLICY_XML,
+        operations = None,
+        tags = tags,
+        productNames = product_names,
+        subscriptionRequired = True
+    )
+    
+    assert api.name == EXAMPLE_NAME
+    assert api.displayName == EXAMPLE_DISPLAY_NAME
+    assert api.path == EXAMPLE_PATH
+    assert api.description == EXAMPLE_DESCRIPTION
+    assert api.policyXml == EXAMPLE_POLICY_XML
+    assert api.operations == []
+    assert api.tags == tags
+    assert api.productNames == product_names
+    assert api.subscriptionRequired == True
+    
+    d = api.to_dict()
+    assert d["name"] == EXAMPLE_NAME
+    assert d["displayName"] == EXAMPLE_DISPLAY_NAME
+    assert d["path"] == EXAMPLE_PATH
+    assert d["description"] == EXAMPLE_DESCRIPTION
+    assert d["policyXml"] == EXAMPLE_POLICY_XML
+    assert d["tags"] == tags
+    assert d["productNames"] == product_names
+    assert d["subscriptionRequired"] == True
+
+
+# ------------------------------