feat: add IaC state store versioning and access hardening

platform/terraform/modules/cloud_azure/modules/aks_rbac/outputs.tf

@@ -2,8 +2,3 @@ output "app_client_id" {
   value       = azuread_application.appname.application_id
   description = "ID of current app client"
 }
-
-output "subscription_id" {
-  value       = data.azurerm_client_config.current_subscription.subscription_id
-  description = "Current azure subscription ID"
-}

tools/cli/commands/setup.py

@@ -771,6 +771,10 @@ def setup(
     else:
         click.echo("12/12: Skipped core services configuration.")
 
+    # restrict access to IaC remote state store
+    cloud_man.protect_iac_state_storage(p.internals["TF_BACKEND_STORAGE_NAME"],
+                                        p.parameters["<IAC_PR_AUTOMATION_IAM_ROLE_RN>"])
+
     show_credentials(p)
 
     return True

tools/cli/services/cloud/aws/aws_manager.py

@@ -36,15 +36,26 @@ def create_iac_state_storage(self, name: str, **kwargs: dict) -> str:
         if kwargs and "region" in kwargs:
             region = kwargs["region"]
         tf_backend_storage_name = f'{name}-{random_string_generator()}'.lower()
+
         self.__aws_sdk.create_bucket(tf_backend_storage_name, region)
+        self.__aws_sdk.enable_bucket_versioning(tf_backend_storage_name, region)
+
         return tf_backend_storage_name
 
     @trace()
-    def destroy_iac_state_storage(self, bucket: str) -> bool:
+    def protect_iac_state_storage(self, name: str, identity: str, **kwargs: dict):
+        region = self.region
+        if kwargs and "region" in kwargs:
+            region = kwargs["region"]
+
+        self.__aws_sdk.set_bucket_policy(name, identity, region)
+
+    @trace()
+    def destroy_iac_state_storage(self, name: str) -> bool:
         """
         Destroy cloud native terraform remote state storage
         """
-        return self.__aws_sdk.delete_bucket(bucket)
+        return self.__aws_sdk.delete_bucket(name)
 
     @trace()
     def create_iac_backend_snippet(self, location: str, service: str, **kwargs: dict) -> str:

tools/cli/services/cloud/aws/aws_sdk.py

@@ -1,7 +1,9 @@
+import json
 import time
 from datetime import datetime, timedelta
 from typing import Dict, List, Optional, Tuple
 
+import boto3
 from awscli.customizations.eks.get_token import STSClientFactory, TokenGenerator, TOKEN_EXPIRATION_MINS
 from botocore.exceptions import ClientError
 
@@ -111,6 +113,45 @@ def create_bucket(self, bucket_name, region=None) -> str:
             return False
         return bucket_name
 
+    def enable_bucket_versioning(self, bucket_name, region=None):
+        if region is None:
+            region = self.region
+
+        resource = boto3.resource("s3", region_name=region)
+        versioning = resource.BucketVersioning(bucket_name)
+        versioning.enable()
+
+    def set_bucket_policy(self, bucket_name: str, identity: str, region: str = None):
+        if region is None:
+            region = self.region
+
+        bucket_policy = {
+            "Version": "2012-10-17",
+            "Statement": [
+                {
+                    "Sid": "RestrictS3Access",
+                    "Principal": {
+                        "AWS": [
+                            self.current_user_arn(),
+                            identity
+                        ]
+                    },
+                    "Effect": "Allow",
+                    "Action": ["s3:*"],
+                    "Resource": [f"arn:aws:s3:::{bucket_name}"]
+                }
+            ]
+        }
+
+        policy_string = json.dumps(bucket_policy)
+
+        s3_client = self._session_manager.session.client('s3', region_name=region)
+
+        s3_client.put_bucket_policy(
+            Bucket=bucket_name,
+            Policy=policy_string
+        )
+
     def get_name_servers(self, domain_name: str) -> Tuple[List[str], str, bool]:
         r53_client = self._session_manager.session.client('route53')
         hosted_zones = r53_client.list_hosted_zones()

tools/cli/services/cloud/azure/azure_manager.py

@@ -24,6 +24,16 @@ def __init__(self, subscription_id: str, location: Optional[str] = None):
     def region(self):
         return self.__azure_sdk.location
 
+    @trace()
+    def protect_iac_state_storage(self, name: str, identity: str):
+        """
+        Restrict access to cloud native terraform remote state storage
+        """
+        self.iac_backend_storage_container_name = name
+        resource_group_name = self._generate_resource_group_name()
+        storage_account_name = self._generate_storage_account_name()
+        self.__azure_sdk.set_storage_access(identity, storage_account_name, resource_group_name)
+
     @trace()
     def destroy_iac_state_storage(self, bucket: str) -> bool:
         """
@@ -111,9 +121,15 @@ def create_iac_state_storage(self, name: str, **kwargs: dict) -> str:
         :return: Resource identifier
         """
         self.iac_backend_storage_container_name = f"{name}-{random_string_generator()}".lower()
-        return self.__azure_sdk.create_storage(self.iac_backend_storage_container_name,
-                                               self._generate_storage_account_name(),
-                                               self._generate_resource_group_name())
+
+        resource_group_name = self._generate_resource_group_name()
+        storage_account_name = self._generate_storage_account_name()
+        storage = self.__azure_sdk.create_storage(self.iac_backend_storage_container_name,
+                                                  storage_account_name,
+                                                  resource_group_name)
+        self.__azure_sdk.set_storage_account_versioning(storage_account_name, resource_group_name)
+
+        return storage
 
     @trace()
     def evaluate_permissions(self) -> bool:

tools/cli/services/cloud/azure/azure_sdk.py

@@ -1,4 +1,5 @@
 import time
+import uuid
 from typing import List, Tuple, Optional
 
 from azure.core.exceptions import ResourceNotFoundError, HttpResponseError, AzureError, ResourceExistsError
@@ -377,6 +378,37 @@ def create_storage_account(self, resource_group_name: str, storage_account_name:
         except ResourceExistsError:
             logger.warning(f"Storage name {storage_account_name} is already in use. Try another name.")
 
+    def set_storage_account_versioning(self, storage_account_name: str, resource_group_name: str) -> None:
+        """
+        Set a storage account data protection options.
+        """
+        try:
+            self.storage_mgmt_client.blob_services.set_service_properties(resource_group_name, storage_account_name,
+                                                                          {
+                                                                              "is_versioning_enabled": True,
+                                                                              "delete_retention_policy": {
+                                                                                  "additional_properties": {},
+                                                                                  "enabled": True,
+                                                                                  "days": 7,
+                                                                                  "allow_permanent_delete": False
+                                                                              }
+                                                                          })
+        except Exception as e:
+            logger.warning(f"Error while setting blob storage versioning {e}")
+
+        logger.info(f"Set storage account {storage_account_name} data versioning options")
+
+    def set_storage_access(self, identity: str, storage_account_name: str, resource_group_name: str):
+        scope = f"subscriptions/{self.subscription_id}/resourcegroups/{resource_group_name}/providers/Microsoft.Storage/storageAccounts/{storage_account_name}"
+        response = self.authorization_client.role_assignments.create(scope,
+                                                                     uuid.uuid4(),
+                                                                     {
+                                                                         # Storage Blob Data Owner
+                                                                         "role_definition_id": "/providers/Microsoft.Authorization/roleDefinitions/b7e6dc6d-f1e8-4753-8033-0f276bb0955b",
+                                                                         "principal_id": identity
+                                                                     })
+        return response
+
     def create_blob_container(self, storage_account_name: str, container_name: str) -> None:
         """Create a blob container in the specified storage account.
 

tools/cli/services/cloud/cloud_provider_manager.py

@@ -32,6 +32,13 @@ def create_iac_state_storage(self, name: str, **kwargs: dict) -> str:
         """
         pass
 
+    @abstractmethod
+    def protect_iac_state_storage(self, name: str, identity: str):
+        """
+        Restrict access to cloud native terraform remote state storage
+        """
+        pass
+
     @abstractmethod
     def destroy_iac_state_storage(self, bucket: str):
         """