Create the Python Track 2 SDK for the Microsoft Azure Attestation Service.

sdk/attestation/azure-security-attestation/SWAGGER.md

@@ -0,0 +1,38 @@
+# Azure Mixed Reality Authentication Service client library for Python
+
+## Setup
+
+```ps
+npm install -g autorest
+```
+
+## Generation
+
+```ps
+cd <swagger-folder>
+autorest SWAGGER.md
+```
+
+### Code generation settings
+
+```yaml
+title: AzureAttestationRestClient
+require: 
+  - https://raw.githubusercontent.com/Azure/azure-rest-api-specs/master/specification/attestation/data-plane/readme.md
+output-folder: azure/security/attestation/_generated
+namespace: azure.security.attestation._generated
+no-namespace-folders: true
+license-header: MICROSOFT_MIT_NO_VERSION
+tag: package-2020-10-01
+package-version: 1.0.0b2
+enable-xml: false
+clear-output-folder: true
+python: true
+v3: true
+add-credentials: true
+azure-arm: false
+payload-flattening-threshold: 2
+package-name: azure-attestation
+credential-scopes: 'https://attest.azure.net/.default'
+
+```

sdk/attestation/azure-security-attestation/azure/security/attestation/__init__.py

@@ -6,11 +6,24 @@
 # Changes may cause incorrect behavior and will be lost if the code is regenerated.
 # --------------------------------------------------------------------------
 
-from ._attestation_client import AttestationClient
+from ._client import AttestationClient
+from ._administration_client import AttestationAdministrationClient
+from ._models import AttestationResult, AttestationSigner, AttestationToken, SigningKey
+from ._generated.models import AttestationType
+from ._configuration import TokenValidationOptions
 from ._version import VERSION
 
 __version__ = VERSION
-__all__ = ['AttestationClient']
+__all__ = [
+    'AttestationClient',
+    'AttestationAdministrationClient',
+    'AttestationType',
+    'AttestationToken',
+    'AttestationSigner',
+    'AttestationResult',
+    'TokenValidationOptions',
+    'SigningKey',
+]
 
 try:
     from ._patch import patch_sdk  # type: ignore

sdk/attestation/azure-security-attestation/azure/security/attestation/_administration_client.py

@@ -0,0 +1,122 @@
+# coding=utf-8
+# --------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# Code generated by Microsoft (R) AutoRest Code Generator.
+# Changes may cause incorrect behavior and will be lost if the code is regenerated.
+# --------------------------------------------------------------------------
+
+from typing import TYPE_CHECKING
+
+from azure.core import PipelineClient
+from msrest import Deserializer, Serializer
+from six import python_2_unicode_compatible
+
+if TYPE_CHECKING:
+    # pylint: disable=unused-import,ungrouped-imports
+    from typing import Any
+
+    from azure.core.credentials import TokenCredential
+    from azure.core.pipeline.transport import HttpRequest, HttpResponse
+
+from ._generated import AzureAttestationRestClient
+from ._generated.models import AttestationType, PolicyResult, StoredAttestationPolicy
+from ._configuration import AttestationClientConfiguration
+from ._models import AttestationSigner, AttestationToken, AttestationResult, StoredAttestationPolicy
+from ._common import Base64Url
+import cryptography
+import cryptography.x509
+import base64
+from typing import List, Any
+from azure.core.tracing.decorator import distributed_trace
+from threading import Lock, Thread
+
+
+class AttestationAdministrationClient(object):
+    """Describes the interface for the per-tenant enclave service.
+    :param str base_url: base url of the service
+    :param credential: An object which can provide secrets for the attestation service
+    :type credential: azure.TokenCredentials or azure.AsyncTokenCredentials
+    :keyword Pipeline pipeline: If omitted, the standard pipeline is used.
+    :keyword HttpTransport transport: If omitted, the standard pipeline is used.
+    :keyword list[HTTPPolicy] policies: If omitted, the standard pipeline is used.
+    """
+
+    def __init__(
+        self,
+        credential,  # type: "TokenCredential"
+        instance_url,  # type: str
+        **kwargs  # type: Any
+    ):
+        # type: (str, Any, dict) -> None
+        base_url = '{instanceUrl}'
+        if not credential:
+            raise ValueError("Missing credential.")
+        self._config = AttestationClientConfiguration(credential, instance_url, **kwargs)
+        self._client = AzureAttestationRestClient(credential, instance_url, **kwargs)
+        self._statelock = Lock()
+        self._signing_certificates = None
+
+    @distributed_trace
+    def get_policy(self, attestation_type: AttestationType) -> AttestationResult[str]:
+        """ Retrieves the attestation policy for a specified attestation type.
+        :param attestation_type - The attestation parameter type.
+        :type attestation_type: AttestationType
+        """
+
+        policyResult = self._client.policy.get(attestation_type)
+        token = AttestationToken[PolicyResult](token=policyResult.token)
+        token_body = token.get_body()
+        stored_policy = AttestationToken[StoredAttestationPolicy](token=token_body['x-ms-policy'])
+
+        actual_policy = Base64Url.decode(stored_policy.get_body()['AttestationPolicy']).decode()
+
+        if self._config.token_validation_options.validate_token:
+            token.validate_token(self._config.token_validation_options, self._get_signers())
+
+        return AttestationResult[str](token, actual_policy)
+
+    @distributed_trace
+    def set_policy(self, attestation_type: AttestationType, attestation_policy: str, signing_key=None) -> AttestationResult[PolicyResult]:
+        base64_policy = Base64Url.encode(attestation_policy.encode('utf-8'))
+        policy_token = AttestationToken[StoredAttestationPolicy](body={"AttestationPolicy": base64_policy})
+        policyResult = self._client.policy.set(attestation_type=attestation_type, new_attestation_policy=policy_token.serialize())
+        token = AttestationToken[PolicyResult](token=policyResult.token)
+        if self._config.token_validation_options.validate_token:
+            token.validate_token(self._config.token_validation_options, self._get_signers())
+
+        return AttestationResult[str](token, token.get_body())
+
+
+    @distributed_trace
+    def _get_signers(self) -> List[AttestationSigner]:
+        """ Returns the set of signing certificates used to sign attestation tokens.
+        """
+
+        with self._statelock:
+            if (self._signing_certificates == None):
+                signing_certificates = self._client.signing_certificates.get()
+                self._signing_certificates = []
+                for key in signing_certificates.keys:
+                    # Convert the returned certificate chain into an array of X.509 Certificates.
+                    certificates = []
+                    for x5c in key.x5_c:
+                        der_cert = base64.b64decode(x5c)
+                        cert = cryptography.x509.load_der_x509_certificate(der_cert)
+                        certificates.append(cert)
+                    self._signing_certificates.append(AttestationSigner(certificates, key.kid))
+            signers = self._signing_certificates
+        return signers
+
+    def close(self):
+        # type: () -> None
+        self._client.close()
+
+    def __enter__(self):
+        # type: () -> AttestationAdministrationClient
+        self._client.__enter__()
+        return self
+
+    def __exit__(self, *exc_details):
+        # type: (Any) -> None
+        self._client.__exit__(*exc_details)

sdk/attestation/azure-security-attestation/azure/security/attestation/_client.py

@@ -0,0 +1,90 @@
+# coding=utf-8
+# --------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# Code generated by Microsoft (R) AutoRest Code Generator.
+# Changes may cause incorrect behavior and will be lost if the code is regenerated.
+# --------------------------------------------------------------------------
+
+from typing import TYPE_CHECKING
+
+from azure.core import PipelineClient
+from msrest import Deserializer, Serializer
+
+if TYPE_CHECKING:
+    # pylint: disable=unused-import,ungrouped-imports
+    from typing import Any
+
+    from azure.core.credentials import TokenCredential
+    from azure.core.pipeline.transport import HttpRequest, HttpResponse
+
+from ._generated import AzureAttestationRestClient
+from ._configuration import AttestationClientConfiguration
+from ._models import AttestationSigner
+import base64
+import cryptography
+import cryptography.x509
+from typing import List, Any
+from azure.core.tracing.decorator import distributed_trace
+
+
+class AttestationClient(object):
+    """Describes the interface for the per-tenant enclave service.
+    :param str base_url: base url of the service
+    :param credential: An object which can provide secrets for the attestation service
+    :type credential: azure.TokenCredentials or azure.AsyncTokenCredentials
+    :keyword Pipeline pipeline: If omitted, the standard pipeline is used.
+    :keyword HttpTransport transport: If omitted, the standard pipeline is used.
+    :keyword list[HTTPPolicy] policies: If omitted, the standard pipeline is used.
+    """
+
+    def __init__(
+        self,
+        credential,  # type: "TokenCredential"
+        instance_url,  # type: str
+        **kwargs  # type: Any
+    ):
+        # type: (str, Any, dict) -> None
+        base_url = '{instanceUrl}'
+        if not credential:
+            raise ValueError("Missing credential.")
+        self._config = AttestationClientConfiguration(credential, instance_url, **kwargs)
+        self._client = AzureAttestationRestClient(credential, instance_url, **kwargs)
+
+    @distributed_trace
+    def get_openidmetadata(self):
+        """ Retrieves the OpenID metadata configuration document for this attestation instance.
+        """
+        return self._client.metadata_configuration.get()
+
+    @distributed_trace
+    def get_signing_certificates(self): # type: () ->List[AttestationSigner]
+        """ Returns the set of signing certificates used to sign attestation tokens.
+        """
+        signing_certificates = self._client.signing_certificates.get()
+        assert signing_certificates.keys is not None
+        signers = []
+        for key in signing_certificates.keys:
+            assert key.x5_c is not None
+
+            # Convert the returned certificate chain into an array of X.509 Certificates.
+            certificates = []
+            for x5c in key.x5_c:
+                der_cert = base64.b64decode(x5c)
+                cert = cryptography.x509.load_der_x509_certificate(der_cert)
+                certificates.append(cert)
+            signers.append(AttestationSigner(certificates, key.kid))
+        return signers
+
+    def close(self):
+        # type: () -> None
+        self._client.close()
+
+    def __enter__(self):
+        # type: () -> AttestationClient
+        self._client.__enter__()
+        return self
+
+    def __exit__(self, *exc_details):
+        # type: (Any) -> None
+        self._client.__exit__(*exc_details)

sdk/attestation/azure-security-attestation/azure/security/attestation/_common.py

@@ -0,0 +1,29 @@
+# coding=utf-8
+# --------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# Code generated by Microsoft (R) AutoRest Code Generator.
+# Changes may cause incorrect behavior and will be lost if the code is regenerated.
+# --------------------------------------------------------------------------
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    # pylint: disable=unused-import,ungrouped-imports
+    from typing import Any
+
+
+import base64
+
+class Base64Url:
+    """Equivalent to base64.urlsafe_b64encode, but strips padding from the encoded and decoded strings.
+    """
+    @staticmethod
+    def encode(unencoded):
+        base64val= base64.urlsafe_b64encode(unencoded)
+        strip_trailing=base64val.split(b'=')[0] # pick the string before the trailing =
+        return str(strip_trailing, 'utf-8')
+
+    @staticmethod
+    def decode(encoded):
+        padding_added = encoded + "=" * ((len(encoded)* -1) % 4)
+        return base64.urlsafe_b64decode(padding_added)

sdk/attestation/azure-security-attestation/azure/security/attestation/_configuration.py

@@ -6,17 +6,17 @@
 # Changes may cause incorrect behavior and will be lost if the code is regenerated.
 # --------------------------------------------------------------------------
 
-from typing import TYPE_CHECKING
+from typing import NoReturn, TYPE_CHECKING
 
 from azure.core.configuration import Configuration
 from azure.core.pipeline import policies
 
 from ._version import VERSION
+from ._models import TokenValidationOptions
 
 if TYPE_CHECKING:
     # pylint: disable=unused-import,ungrouped-imports
     from typing import Any
-
     from azure.core.credentials import TokenCredential
 
 
@@ -30,6 +30,8 @@ class AttestationClientConfiguration(Configuration):
     :type credential: ~azure.core.credentials.TokenCredential
     :param instance_url: The attestation instance base URI, for example https://mytenant.attest.azure.net.
     :type instance_url: str
+    :keyword validation_options: Optional token validation options.
+    :type validation_options: TokenValidationOptions
     """
 
     def __init__(
@@ -45,11 +47,14 @@ def __init__(
             raise ValueError("Parameter 'instance_url' must not be None.")
         super(AttestationClientConfiguration, self).__init__(**kwargs)
 
+        self.token_validation_options = kwargs.get('validation_options') # type: TokenValidationOptions
+        if (self.token_validation_options == None):
+            self.token_validation_options = TokenValidationOptions(validate_token=True)
         self.credential = credential
         self.instance_url = instance_url
         self.api_version = "2020-10-01"
         self.credential_scopes = kwargs.pop('credential_scopes', ['https://attest.azure.net/.default'])
-        kwargs.setdefault('sdk_moniker', 'security-attestation/{}'.format(VERSION))
+        kwargs.setdefault('sdk_moniker', 'attestation/{}'.format(VERSION))
         self._configure(**kwargs)
 
     def _configure(

sdk/attestation/azure-security-attestation/azure/security/attestation/_generated/__init__.py

@@ -0,0 +1,19 @@
+# coding=utf-8
+# --------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# Code generated by Microsoft (R) AutoRest Code Generator.
+# Changes may cause incorrect behavior and will be lost if the code is regenerated.
+# --------------------------------------------------------------------------
+
+from ._azure_attestation_rest_client import AzureAttestationRestClient
+from ._version import VERSION
+
+__version__ = VERSION
+__all__ = ['AzureAttestationRestClient']
+
+try:
+    from ._patch import patch_sdk  # type: ignore
+    patch_sdk()
+except ImportError:
+    pass