# LocalStack Resource Provider Scaffolding v2
from __future__ import annotations

import json
import logging
import random
import string
from pathlib import Path
from typing import TypedDict

import localstack.services.cloudformation.provider_utils as util
from localstack.services.cloudformation.resource_provider import (
    OperationStatus,
    ProgressEvent,
    ResourceProvider,
    ResourceRequest,
)

LOG = logging.getLogger(__name__)


class SecretsManagerSecretProperties(TypedDict):
    Description: str | None
    GenerateSecretString: GenerateSecretString | None
    Id: str | None
    KmsKeyId: str | None
    Name: str | None
    ReplicaRegions: list[ReplicaRegion] | None
    SecretString: str | None
    Tags: list[Tag] | None


class GenerateSecretString(TypedDict):
    ExcludeCharacters: str | None
    ExcludeLowercase: bool | None
    ExcludeNumbers: bool | None
    ExcludePunctuation: bool | None
    ExcludeUppercase: bool | None
    GenerateStringKey: str | None
    IncludeSpace: bool | None
    PasswordLength: int | None
    RequireEachIncludedType: bool | None
    SecretStringTemplate: str | None


class ReplicaRegion(TypedDict):
    Region: str | None
    KmsKeyId: str | None


class Tag(TypedDict):
    Key: str | None
    Value: str | None


REPEATED_INVOCATION = "repeated_invocation"


class SecretsManagerSecretProvider(ResourceProvider[SecretsManagerSecretProperties]):
    TYPE = "AWS::SecretsManager::Secret"  # Autogenerated. Don't change
    SCHEMA = util.get_schema_path(Path(__file__))  # Autogenerated. Don't change

    def create(
        self,
        request: ResourceRequest[SecretsManagerSecretProperties],
    ) -> ProgressEvent[SecretsManagerSecretProperties]:
        """
        Create a new resource.

        Primary identifier fields:
          - /properties/Id



        Create-only properties:
          - /properties/Name

        Read-only properties:
          - /properties/Id

        IAM permissions required:
          - secretsmanager:DescribeSecret
          - secretsmanager:GetRandomPassword
          - secretsmanager:CreateSecret
          - secretsmanager:TagResource

        """
        model = request.desired_state
        secrets_manager = request.aws_client_factory.secretsmanager

        if not model.get("Name"):
            # not actually correct. Given the LogicalResourceId "MySecret",
            # an example for the generated name would be "MySecret-krxoxgcznYdq-sQNsqO"
            model["Name"] = util.generate_default_name(
                stack_name=request.stack_name, logical_resource_id=request.logical_resource_id
            )

        attributes = ["Name", "Description", "KmsKeyId", "SecretString", "Tags"]
        params = util.select_attributes(model, attributes)

        """
        From CFn Docs:
        If you omit both GenerateSecretString and SecretString, you create an empty secret.
        When you make a change to this property, a new secret version is created.
        CDK wil generate empty dict in which case we also need to generate SecretString
        """

        gen_secret = model.get("GenerateSecretString")
        if gen_secret is not None:
            secret_value = self._get_secret_value(gen_secret)
            template = gen_secret.get("SecretStringTemplate")
            if template:
                secret_value = self._modify_secret_template(template, secret_value, gen_secret)
            params["SecretString"] = secret_value

        response = secrets_manager.create_secret(**params)
        model["Id"] = response["ARN"]

        return ProgressEvent(
            status=OperationStatus.SUCCESS,
            resource_model=model,
            custom_context=request.custom_context,
        )

    def _get_secret_value(self, gen_secret):
        excl_lower = gen_secret.get("ExcludeLowercase")
        excl_upper = gen_secret.get("ExcludeUppercase")
        excl_chars = gen_secret.get("ExcludeCharacters") or ""
        excl_numbers = gen_secret.get("ExcludeNumbers")
        excl_punct = gen_secret.get("ExcludePunctuation")
        incl_spaces = gen_secret.get("IncludeSpace")
        length = gen_secret.get("PasswordLength") or 32
        req_each = gen_secret.get("RequireEachIncludedType")
        return self.generate_secret_value(
            length=length,
            excl_lower=excl_lower,
            excl_upper=excl_upper,
            excl_punct=excl_punct,
            incl_spaces=incl_spaces,
            excl_chars=excl_chars,
            excl_numbers=excl_numbers,
            req_each=req_each,
        )

    def _modify_secret_template(self, template, secret_value, gen_secret):
        gen_key = gen_secret.get("GenerateStringKey") or "secret"
        template = json.loads(template)
        template[gen_key] = secret_value
        return json.dumps(template)

    def generate_secret_value(
        self,
        length: int,
        excl_lower: bool,
        excl_upper: bool,
        excl_chars: str,
        excl_numbers: bool,
        excl_punct: bool,
        incl_spaces: bool,
        req_each: bool,
    ) -> str:
        """WARN: This is NOT a secure way to generate secrets - use only for testing and not in production use cases!"""

        # TODO: add a couple of unit tests for this function ...

        punctuation = r"!\"#$%&'()*+,-./:;<=>?@[\]^_`{|}~"
        alphabet = ""
        if not excl_punct:
            alphabet += punctuation
        if not excl_upper:
            alphabet += string.ascii_uppercase
        if not excl_lower:
            alphabet += string.ascii_lowercase
        if not excl_numbers:
            alphabet += "".join([str(i) for i in list(range(10))])
        if incl_spaces:
            alphabet += " "
        if req_each:
            LOG.info("Secret generation option 'RequireEachIncludedType' not yet supported")

        for char in excl_chars:
            alphabet = alphabet.replace(char, "")

        result = [alphabet[random.randrange(len(alphabet))] for _ in range(length)]
        result = "".join(result)
        return result

    def read(
        self,
        request: ResourceRequest[SecretsManagerSecretProperties],
    ) -> ProgressEvent[SecretsManagerSecretProperties]:
        """
        Fetch resource information

        IAM permissions required:
          - secretsmanager:DescribeSecret
          - secretsmanager:GetSecretValue
        """
        secretsmanager = request.aws_client_factory.secretsmanager
        secret_id = request.desired_state["Id"]

        secret = secretsmanager.describe_secret(SecretId=secret_id)
        model = SecretsManagerSecretProperties(
            **util.select_attributes(secret, self.SCHEMA["properties"])
        )
        model["Id"] = secret["ARN"]

        if "Tags" not in model:
            model["Tags"] = []

        model["ReplicaRegions"] = [
            {"KmsKeyId": replication_region["KmsKeyId"], "Region": replication_region["Region"]}
            for replication_region in secret.get("ReplicationStatus", [])
        ]
        if "ReplicaRegions" not in model:
            model["ReplicaRegions"] = []

        return ProgressEvent(status=OperationStatus.SUCCESS, resource_model=model)

    def delete(
        self,
        request: ResourceRequest[SecretsManagerSecretProperties],
    ) -> ProgressEvent[SecretsManagerSecretProperties]:
        """
        Delete a resource

        IAM permissions required:
          - secretsmanager:DeleteSecret
          - secretsmanager:DescribeSecret
          - secretsmanager:RemoveRegionsFromReplication
        """
        model = request.desired_state
        secrets_manager = request.aws_client_factory.secretsmanager

        secrets_manager.delete_secret(SecretId=model["Name"], ForceDeleteWithoutRecovery=True)

        return ProgressEvent(
            status=OperationStatus.SUCCESS,
            resource_model=model,
            custom_context=request.custom_context,
        )

    def update(
        self,
        request: ResourceRequest[SecretsManagerSecretProperties],
    ) -> ProgressEvent[SecretsManagerSecretProperties]:
        """
        Update a resource

        IAM permissions required:
          - secretsmanager:UpdateSecret
          - secretsmanager:TagResource
          - secretsmanager:UntagResource
          - secretsmanager:GetRandomPassword
          - secretsmanager:GetSecretValue
          - secretsmanager:ReplicateSecretToRegions
          - secretsmanager:RemoveRegionsFromReplication
        """
        raise NotImplementedError

    def list(
        self,
        request: ResourceRequest[SecretsManagerSecretProperties],
    ) -> ProgressEvent[SecretsManagerSecretProperties]:
        resources = request.aws_client_factory.secretsmanager.list_secrets()
        return ProgressEvent(
            status=OperationStatus.SUCCESS,
            resource_models=[
                SecretsManagerSecretProperties(Id=resource["Name"])
                for resource in resources["SecretList"]
            ],
        )
