server.py

from __future__ import annotations

from datetime import date, timedelta
from typing import Dict, Optional, Tuple

from kvac.commitment import BlindAttributeCommitment
from kvac.credential_presentation import CredentialPresentation
from kvac.issuance_request import IssuanceRequest
from kvac.issuance_response import IssuanceResponse
from kvac.issuer_key import IssuerKeyPair, IssuerPublicKey
from kvac.system_params import SystemParams
from kvac.verifiable_encryption import Ciphertext, KeyCommitment
from poksho.group.ristretto import RistrettoScalar

from zkgroup.auth_credential import AuthCredential
from zkgroup.constants import Role, MAX_ATTRIBUTES, REDEMPTION_DATE_RANGE
from zkgroup.exceptions import NotAMemberError, NotAnAdministratorError
from zkgroup.profile_key_credential import ProfileKeyCredential


class Server:
    """Represents the ZKGroup Server. It is a singleton, which means that subsequent calls
    'Server()' will return the same instance. It manages groups and their registered users."""
    _instance: Optional[Server] = None
    _initialized: bool = False

    def __new__(cls):
        if cls._instance is None:
            cls._instance = super(Server, cls).__new__(cls)
            cls._initialized = False
        return cls._instance

    def __init__(self):
        if self._initialized:
            # required check because __init__ can be re-called
            return

        self.system_params = SystemParams.generate(MAX_ATTRIBUTES, "zkgroup.server.server")
        self._auth_keypair = IssuerKeyPair.generate(
            self.system_params,
            AuthCredential.number_of_attribute_components())
        self._profile_key_keypair = IssuerKeyPair.generate(
            self.system_params,
            ProfileKeyCredential.number_of_attribute_components())

        # use tuple of encoded KeyCommitments as group key
        # use encoded UidCiphertext as user key
        # for each user we store an optional encoded ProfileKeyCiphertext and their Role
        self.groups: \
            Dict[Tuple[bytes, bytes],
                Dict[bytes,
                    Tuple[Optional[bytes], Role]]] = {}

        # (uid, profile_key_version) --> profile_key_commitment
        self.profile_key_commitments: Dict[
            Tuple[bytes, bytes], BlindAttributeCommitment] = {}

        self._initialized = True

    # Operations for Credentials
    def get_auth_credential(
            self,
            request: IssuanceRequest,
            redemption_date: date,
    ) -> IssuanceResponse:
        """The server issues an AuthCredential."""
        # assumes an authenticated channel

        # check: authenticated uid == request.clear_attributes[1].to_bytes()
        if self.system_params.G_ms[2] ** RistrettoScalar.from_bytes(
                redemption_date.toordinal().to_bytes(32, byteorder="little")
        ) != request.clear_attributes[2]:
            raise ValueError("Redemption date does not match the one in the request.")
        if redemption_date > date.today() + timedelta(days=REDEMPTION_DATE_RANGE):
            raise ValueError("Redemption date is too far in the future.")

        return AuthCredential.issue(issuer_key=self._auth_keypair, request=request)

    def commit_to_profile_key(
            self,
            uid: bytes,
            profile_key_version: bytes,
            profile_key_commitment: BlindAttributeCommitment
    ):
        """The server stores the profile key commitment of a user."""
        # assumes an authenticated channel
        self.profile_key_commitments[(uid, profile_key_version)] = profile_key_commitment

    def get_profile_key_credential(
            self,
            request: IssuanceRequest,
            uid: bytes,
            profile_key_version: bytes,
            commitment: BlindAttributeCommitment
    ) -> IssuanceResponse:
        """The server issues a ProfileKeyCredential."""
        if (uid, profile_key_version) not in self.profile_key_commitments:
            raise ValueError("Profile key commitment not found.")
        if self.profile_key_commitments[(uid, profile_key_version)] != commitment:
            raise ValueError("Profile key commitment does not match.")

        return ProfileKeyCredential.issue(
            issuer_key=self._profile_key_keypair,
            request=request,
            commitment=commitment)

    def _verify_auth_credential(
            self,
            presentation: CredentialPresentation,
            redemption_date: date,
    ) -> bool:
        """The server verifies an AuthCredential."""
        if self.system_params.G_ms[2] ** RistrettoScalar.from_bytes(
            redemption_date.toordinal().to_bytes(32, byteorder="little")
        ) != presentation.attributes[1].value:
            raise ValueError("Redemption date does not match the one in the presentation.")
        if redemption_date < date.today():
            raise ValueError("Redemption date is in the past.")
        return AuthCredential.verify_presentation(
            issuer_key=self._auth_keypair,
            presentation=presentation)

    def _verify_profile_key_credential(
            self,
            presentation: CredentialPresentation,
    ) -> bool:
        """The server verifies a ProfileKeyCredential."""
        return ProfileKeyCredential.verify_presentation(
            issuer_key=self._profile_key_keypair,
            presentation=presentation)

    # Operations for Group Management
    def auth_as_group_member(
            self,
            group_public_params: Tuple[KeyCommitment, KeyCommitment],
            auth_credential_presentation: CredentialPresentation,
            redemption_date: date,
    ) -> bool:
        if not self._verify_auth_credential(auth_credential_presentation, redemption_date):
            raise ValueError("Invalid auth credential presentation.")
        uid_ciphertext = auth_credential_presentation.attributes[0].value
        return bytes(uid_ciphertext) in self.groups[(bytes(group_public_params[0]), bytes(group_public_params[1]))]

    def add_group_member(
            self,
            group_public_params: Tuple[KeyCommitment, KeyCommitment],
            target_uid_ciphertext: Ciphertext,
            target_profile_key_ciphertext: Ciphertext,
            target_role: Role,
            auth_credential_presentation: CredentialPresentation,
            pk_credential_presentation: CredentialPresentation,
            redemption_date: date,
    ):
        if not self.auth_as_group_member(
                group_public_params,
                auth_credential_presentation,
                redemption_date):
            raise NotAMemberError()

        if not self._verify_profile_key_credential(pk_credential_presentation):
            raise ValueError("Invalid profile key credential presentation")

        group = self.groups[(bytes(group_public_params[0]), bytes(group_public_params[1]))]

        if not self.user_is_admin(
                group_public_params,
                auth_credential_presentation):
            raise NotAnAdministratorError()

        if bytes(target_uid_ciphertext) in group:
            target_user = group[bytes(target_uid_ciphertext)]
            if not target_user[0] is None:
                raise ValueError("User already exists.")

        group[bytes(target_uid_ciphertext)] = (bytes(target_profile_key_ciphertext), target_role)

    def create_group(
            self,
            group_public_params: Tuple[KeyCommitment, KeyCommitment],
            profile_key_ciphertext: Ciphertext,
            auth_credential_presentation: CredentialPresentation,
            pk_credential_presentation: CredentialPresentation,
            redemption_date: date,
    ):
        if not self._verify_auth_credential(auth_credential_presentation, redemption_date):
            raise ValueError("Invalid auth credential presentation.")

        if not self._verify_profile_key_credential(pk_credential_presentation):
            raise ValueError("Invalid profile key credential presentation")

        uid_ciphertext = auth_credential_presentation.attributes[0].value
        self.groups[(bytes(group_public_params[0]), bytes(group_public_params[1]))] = {
            bytes(uid_ciphertext): (bytes(profile_key_ciphertext), Role.ADMINISTRATOR)}

    def user_is_admin(
            self,
            group_public_params: Tuple[KeyCommitment, KeyCommitment],
            auth_credential_presentation: CredentialPresentation,
    ) -> bool:
        uid_ciphertext = auth_credential_presentation.attributes[0].value
        return self.groups[(bytes(group_public_params[0]), bytes(group_public_params[1]))][
            bytes(uid_ciphertext)][1] == Role.ADMINISTRATOR

    def delete_group(
            self,
            group_public_params: Tuple[KeyCommitment, KeyCommitment],
            auth_credential_presentation: CredentialPresentation,
            redemption_date: date,
    ):
        if not self.auth_as_group_member(
                group_public_params,
                auth_credential_presentation,
                redemption_date):
            raise NotAMemberError()

        if not self.user_is_admin(
                group_public_params,
                auth_credential_presentation):
            raise NotAnAdministratorError()

        del self.groups[(bytes(group_public_params[0]), bytes(group_public_params[1]))]

    def remove_member(
            self,
            group_public_params: Tuple[KeyCommitment, KeyCommitment],
            target_uid_ciphertext: Ciphertext,
            auth_credential_presentation: CredentialPresentation,
            redemption_date: date,
    ):
        if not self.auth_as_group_member(
                group_public_params,
                auth_credential_presentation,
                redemption_date):
            raise NotAMemberError()

        group = self.groups[(bytes(group_public_params[0]), bytes(group_public_params[1]))]
        uid_ciphertext = auth_credential_presentation.attributes[0].value

        if not self.user_is_admin(
                group_public_params,
                auth_credential_presentation) \
                and uid_ciphertext != target_uid_ciphertext:
            raise NotAnAdministratorError()
        del group[bytes(target_uid_ciphertext)]

    def add_invited_group_member(
            self,
            group_public_params: Tuple[KeyCommitment, KeyCommitment],
            new_uid_ciphertext: Ciphertext,
            new_role: Role,
            auth_credential_presentation: CredentialPresentation,
            redemption_date: date,
    ):
        if not self.auth_as_group_member(
                group_public_params,
                auth_credential_presentation,
                redemption_date):
            raise NotAMemberError()

        group = self.groups[(bytes(group_public_params[0]), bytes(group_public_params[1]))]

        if not self.user_is_admin(
                group_public_params,
                auth_credential_presentation):
            raise NotAnAdministratorError()
        if bytes(new_uid_ciphertext) in group:
            raise ValueError("User already exists.")
        group[bytes(new_uid_ciphertext)] = (None, new_role)

    def fetch_group_members(
            self,
            group_public_params: Tuple[KeyCommitment, KeyCommitment],
            auth_credential_presentation: CredentialPresentation,
            redemption_date: date,
    ) -> Dict[bytes, Tuple[Optional[bytes], Role]]:
        if not self.auth_as_group_member(
                group_public_params,
                auth_credential_presentation,
                redemption_date):
            raise NotAMemberError()

        return self.groups[(bytes(group_public_params[0]), bytes(group_public_params[1]))]

    def update_profile_key(
            self,
            group_public_params: Tuple[KeyCommitment, KeyCommitment],
            profile_key_ciphertext: Ciphertext,
            auth_credential_presentation: CredentialPresentation,
            pk_credential_presentation: CredentialPresentation,
            redemption_date: date,
    ):
        if not self.auth_as_group_member(
                group_public_params,
                auth_credential_presentation,
                redemption_date):
            raise NotAMemberError()

        if not self._verify_profile_key_credential(pk_credential_presentation):
            raise ValueError("Invalid profile key credential presentation")

        group = self.groups[(bytes(group_public_params[0]), bytes(group_public_params[1]))]
        uid_ciphertext = auth_credential_presentation.attributes[0].value
        role = group[bytes(uid_ciphertext)][1]

        group[bytes(uid_ciphertext)] = (bytes(profile_key_ciphertext), role)

    @property
    def auth_public_key(self) -> IssuerPublicKey:
        return self._auth_keypair.public

    @property
    def profile_key_public_key(self) -> IssuerPublicKey:
        return self._profile_key_keypair.public