class ActionInProgress(EsteidError):
    """
    Raised when a request to the service is still in pending status,
    or when it is necessary to move on to the next step while passing some data on to the user
    (like after successful auth, we initiate a signing request and display verification code)
    """

    status = HTTPStatus.ACCEPTED

    default_message = _("Operation is in progress.")

    def __init__(self, message=None, data: dict = None):
        super().__init__(message)
        self.data = data or {} class AlreadySignedByUser(InvalidParameters):
    """The container has been already signed by the same party.

    The class is based on InvalidParameters so that such errors are not logged.
    """

    status = HTTPStatus.CONFLICT

    default_message = _("The container has been already signed by the user.") class CanceledByUser(InvalidParameters):
    """User canceled the operation."""

    status = HTTPStatus.CONFLICT

    default_message = _("The signing operation was canceled by the user.") class EsteidError(Exception):
    """
    A generic Esteid error.

    Provides an interface for displaying a translated message to the user.

    Note: by default, any message passed as a first positional argument is ignored, because it only
    serves for better logging. The User Interface should display a verbose error message
    in case of a user-caused error so that user can take action accordingly. Errors caused by service malfunction
    are likely not informative to a user without malicious intent.

    If the default message contains {key} placeholders, the exception is supposed to be
    raised with kwargs, these will be passed on to <translated default message>.format(**kwargs).
    """

    status = HTTPStatus.INTERNAL_SERVER_ERROR

    default_message = _("Something went wrong")

    kwargs: dict

    def __init__(self, message=None, **kwargs):
        super().__init__(message)
        self.kwargs = kwargs

    def get_message(self):
        return str(self.default_message).format(**self.kwargs)

    def get_user_error(self):
        return {
            "error": self.__class__.__name__,  # error code that can be bound to JS translations
            # TODO review errors and add translations
            "message": self.get_message(),
        } class InvalidParameters(EsteidError):
    """Invalid initialization parameters received from the request."""

    status = HTTPStatus.BAD_REQUEST

    default_message = _("Invalid parameters.") class OfflineError(UpstreamServiceError):
    default_message = _("The {service} service is unavailable at the moment. Please try again later.") class UpstreamServiceError(EsteidError):
    """A failure reported by upstream service which is not caused by user.

    Takes a `service` kwarg, e.g. Mobile ID / Smart ID
    """

    status = HTTPStatus.SERVICE_UNAVAILABLE

    default_message = _("The {service} service reported a failure.")

    def __init__(self, message=None, *, service, **kwargs):
        super().__init__(message)
        self.kwargs = {"service": service, **kwargs} class UserNotRegistered(InvalidParameters):
    """User not registered in the service."""

    status = HTTPStatus.CONFLICT

    default_message = _("User not registered in the service.") class Signer(FromDictMixin):
    id_code = attr.ib()

    # Note: This should also support common_name argument coming in
    full_name = attr.ib()

    certificate: Certificate = attr.ib(
        validator=attr.validators.instance_of(Certificate), converter=get_instance_converter(Certificate)
    )

    @staticmethod
    def prepare_kwargs(kwargs):
        kwargs = camel_2_py(kwargs)
        full_name = kwargs.get("full_name", None)

        # If the full_name is None, check common_name field and use that
        if full_name is None:
            full_name = get_name_from_legacy_common_name(kwargs.pop("common_name", None))

        kwargs["full_name"] = full_name

        return kwargs

    @classmethod
    def from_certificate(cls, cert: "Union[bytes, Asn1CryptoCertificate, OsCryptoCertificate]"):
        """
        Get personal info from an oscrypto/asn1crypto Certificate object

        For a closer look at where the attributes come from:
        asn1crypto.x509.NameType
        """
        cert_holder_info = CertificateHolderInfo.from_certificate(cert)
        return cls(
            certificate=Certificate.from_certificate(cert_holder_info.asn1_certificate),
            full_name=f"{cert_holder_info.given_name} {cert_holder_info.surname}",
            id_code=cert_holder_info.id_code,
        ) class SignViewMixin(SessionViewMixin):
    # these come from the `url()` definition as in `View.as_view(signing_method='...')`, either one is enough
    signer_class: Type[Signer] = None
    signing_method: str = None

    def get_container(self, *args, **kwargs) -> Union[str, BinaryIO, pyasice.Container]:
        """
        Returns [path to|file handle of] the container to sign, if it exists prior to signing

        A string value is treated as a file path.
        A file handle is anything that provides a `read()` method that returns bytes.
        """
        raise NotImplementedError

    def save_container(self, container: pyasice.Container, *args, **kwargs):
        """
        Receives a container instance, expected to save the file contents as appropriate.

        Example with Django's UploadedFile:

            from esteid.compat import container_info
            instance = self.get_object()
            # Be sure to call `container_info(container)` before `container.finalize()`
            instance.container_info = container_info(container)
            instance.container = UploadedFile(container.finalize(), "signed_document.doc", container.MIME_TYPE)
            # OR:
            # buffer = container.finalize().getbuffer()
            # instance.container = SimpleUploadedFile("signed_document.doc", buffer, container.MIME_TYPE)
            instance.save()

        """
        raise NotImplementedError

    def get_files_to_sign(self, *args, **kwargs) -> list:
        """
        Returns list of files to sign, unless there is a pre-created container
        """
        raise NotImplementedError

    def get_success_response(self, *args, **kwargs):
        """
        Generates a success response when the signing process is complete (in `finalize()`)

        Can be customized to return additional data, such as links to download the container
        """
        return JsonResponse({"status": self.Status.SUCCESS})

    def check_eligibility(self, signer: Signer, container: pyasice.Container = None):
        """
        Performs a check whether a signing party is eligible to sign the container.

        Override it in subclasses as necessary:
        """
        if container is None or settings.ESTEID_ALLOW_ONE_PARTY_SIGN_TWICE:
            return

        signatories = []
        for signature in container.iter_signatures():
            subject_cert = signature.get_certificate()
            signatory = SignerData.from_certificate(subject_cert)
            signatories.append(signatory.id_code)

        if signer.id_code in signatories:
            raise AlreadySignedByUser

    def select_signer_class(self) -> Type["Signer"]:
        if self.signer_class is not None:
            return self.signer_class
        return Signer.select_signer(self.signing_method)

    def start_session(self, request: "RequestType", *args, **kwargs):
        """
        Initiates a signing session
        """

        signer_class = self.select_signer_class()
        signer = signer_class.start_session(request.session, request.data)

        try:
            container = self.get_container(*args, **kwargs)
        except NotImplementedError:
            container = None

        if not container:
            files_to_sign = self.get_files_to_sign(*args, **kwargs)
        else:
            files_to_sign = None
            if not isinstance(container, pyasice.Container):
                if isinstance(container, str):
                    container = pyasice.Container.open(container)
                else:
                    container = pyasice.Container(container)

        self.check_eligibility(signer, container)

        response_to_user = signer.prepare(container, files_to_sign)

        return JsonResponse({**response_to_user, "status": self.Status.SUCCESS})

    def finish_session(self, request: "RequestType", *args, **kwargs):
        """
        Checks the status of a signing session and attempts to finalize signing
        """
        signer_class = self.select_signer_class()
        signer = signer_class.load_session(request.session)

        do_cleanup = True

        try:
            container = signer.finalize(getattr(request, "data", None))
            self.save_container(container, *args, **kwargs)

        except ActionInProgress as e:
            do_cleanup = False
            return JsonResponse({"status": self.Status.PENDING, **e.data}, status=e.status)

        finally:
            if do_cleanup:
                signer.cleanup()

        return self.get_success_response(*args, **kwargs)
import json
import pytest
from http import HTTPStatus
from unittest.mock import Mock, patch
from pyasice import Container
from esteid.exceptions import (
    ActionInProgress,
    AlreadySignedByUser,
    CanceledByUser,
    EsteidError,
    InvalidParameters,
    OfflineError,
    UpstreamServiceError,
    UserNotRegistered,
)
from esteid.types import Signer as SignerData
from ..views import SignViewMixin






@pytest.fixture()
def signatory_id_code(signed_container):
    signature = next(signed_container.iter_signatures())
    subject_cert = signature.get_certificate()
    signatory = SignerData.from_certificate(subject_cert)
    return signatory.id_code


@pytest.fixture()
def signer_class():
    signer_class = Mock(name="signer class")
    signer_class.start_session.return_value = signer_class()
    signer_class.load_session.return_value = signer_class()
    return signer_class


@pytest.fixture()
def signing_view(signer_class):

    the_view = SignViewMixin()