xref: /aosp_15_r20/external/tink/testing/python/services.py (revision e7b1675dde1b92d52ec075b0a92829627f2c52a5)

# Copyright 2020 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS-IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Testing service API implementations in Python."""

import io

import grpc
import tink
from tink import aead
from tink import cleartext_keyset_handle
from tink import daead
from tink import hybrid
from tink import jwt
from tink import mac
from tink import prf
from tink import signature
from tink import streaming_aead
from tink.proto import tink_pb2
from tink.testing import bytes_io
from protos import testing_api_pb2
from protos import testing_api_pb2_grpc


# All KeyTemplate (as Protobuf) defined in the Python API.
_KEY_TEMPLATE = {
    'AES128_EAX':
        aead.aead_key_templates.AES128_EAX,
    'AES128_EAX_RAW':
        aead.aead_key_templates.AES128_EAX_RAW,
    'AES256_EAX':
        aead.aead_key_templates.AES256_EAX,
    'AES256_EAX_RAW':
        aead.aead_key_templates.AES256_EAX_RAW,
    'AES128_GCM':
        aead.aead_key_templates.AES128_GCM,
    'AES128_GCM_RAW':
        aead.aead_key_templates.AES128_GCM_RAW,
    'AES256_GCM':
        aead.aead_key_templates.AES256_GCM,
    'AES256_GCM_RAW':
        aead.aead_key_templates.AES256_GCM_RAW,
    'AES128_GCM_SIV':
        aead.aead_key_templates.AES128_GCM_SIV,
    'AES128_GCM_SIV_RAW':
        aead.aead_key_templates.AES128_GCM_SIV_RAW,
    'AES256_GCM_SIV':
        aead.aead_key_templates.AES256_GCM_SIV,
    'AES256_GCM_SIV_RAW':
        aead.aead_key_templates.AES256_GCM_SIV_RAW,
    'AES128_CTR_HMAC_SHA256':
        aead.aead_key_templates.AES128_CTR_HMAC_SHA256,
    'AES128_CTR_HMAC_SHA256_RAW':
        aead.aead_key_templates.AES128_CTR_HMAC_SHA256_RAW,
    'AES256_CTR_HMAC_SHA256':
        aead.aead_key_templates.AES256_CTR_HMAC_SHA256,
    'AES256_CTR_HMAC_SHA256_RAW':
        aead.aead_key_templates.AES256_CTR_HMAC_SHA256_RAW,
    'XCHACHA20_POLY1305':
        aead.aead_key_templates.XCHACHA20_POLY1305,
    'XCHACHA20_POLY1305_RAW':
        aead.aead_key_templates.XCHACHA20_POLY1305_RAW,
    'AES256_SIV':
        daead.deterministic_aead_key_templates.AES256_SIV,
    'AES128_CTR_HMAC_SHA256_4KB':
        streaming_aead.streaming_aead_key_templates.AES128_CTR_HMAC_SHA256_4KB,
    'AES128_CTR_HMAC_SHA256_1MB':
        streaming_aead.streaming_aead_key_templates.AES128_CTR_HMAC_SHA256_1MB,
    'AES256_CTR_HMAC_SHA256_4KB':
        streaming_aead.streaming_aead_key_templates.AES256_CTR_HMAC_SHA256_4KB,
    'AES256_CTR_HMAC_SHA256_1MB':
        streaming_aead.streaming_aead_key_templates.AES256_CTR_HMAC_SHA256_1MB,
    'AES128_GCM_HKDF_4KB':
        streaming_aead.streaming_aead_key_templates.AES128_GCM_HKDF_4KB,
    'AES128_GCM_HKDF_1MB':
        streaming_aead.streaming_aead_key_templates.AES128_GCM_HKDF_1MB,
    'AES256_GCM_HKDF_4KB':
        streaming_aead.streaming_aead_key_templates.AES256_GCM_HKDF_4KB,
    'AES256_GCM_HKDF_1MB':
        streaming_aead.streaming_aead_key_templates.AES256_GCM_HKDF_1MB,
    'ECIES_P256_HKDF_HMAC_SHA256_AES128_GCM':
        hybrid.hybrid_key_templates.ECIES_P256_HKDF_HMAC_SHA256_AES128_GCM,
    'ECIES_P256_COMPRESSED_HKDF_HMAC_SHA256_AES128_GCM':
        hybrid.hybrid_key_templates
        .ECIES_P256_COMPRESSED_HKDF_HMAC_SHA256_AES128_GCM,
    'ECIES_P256_HKDF_HMAC_SHA256_AES128_CTR_HMAC_SHA256':
        hybrid.hybrid_key_templates
        .ECIES_P256_HKDF_HMAC_SHA256_AES128_CTR_HMAC_SHA256,
    'ECIES_P256_COMPRESSED_HKDF_HMAC_SHA256_AES128_CTR_HMAC_SHA256':
        hybrid.hybrid_key_templates
        .ECIES_P256_COMPRESSED_HKDF_HMAC_SHA256_AES128_CTR_HMAC_SHA256,
    'DHKEM_X25519_HKDF_SHA256_HKDF_SHA256_AES_128_GCM':
        hybrid.hybrid_key_templates
        .DHKEM_X25519_HKDF_SHA256_HKDF_SHA256_AES_128_GCM,
    'DHKEM_X25519_HKDF_SHA256_HKDF_SHA256_AES_128_GCM_RAW':
        hybrid.hybrid_key_templates
        .DHKEM_X25519_HKDF_SHA256_HKDF_SHA256_AES_128_GCM_RAW,
    'DHKEM_X25519_HKDF_SHA256_HKDF_SHA256_AES_256_GCM':
        hybrid.hybrid_key_templates
        .DHKEM_X25519_HKDF_SHA256_HKDF_SHA256_AES_256_GCM,
    'DHKEM_X25519_HKDF_SHA256_HKDF_SHA256_AES_256_GCM_RAW':
        hybrid.hybrid_key_templates
        .DHKEM_X25519_HKDF_SHA256_HKDF_SHA256_AES_256_GCM_RAW,
    'DHKEM_X25519_HKDF_SHA256_HKDF_SHA256_CHACHA20_POLY1305':
        hybrid.hybrid_key_templates
        .DHKEM_X25519_HKDF_SHA256_HKDF_SHA256_CHACHA20_POLY1305,
    'DHKEM_X25519_HKDF_SHA256_HKDF_SHA256_CHACHA20_POLY1305_RAW':
        hybrid.hybrid_key_templates
        .DHKEM_X25519_HKDF_SHA256_HKDF_SHA256_CHACHA20_POLY1305_RAW,
    'AES_CMAC':
        mac.mac_key_templates.AES_CMAC,
    'HMAC_SHA256_128BITTAG':
        mac.mac_key_templates.HMAC_SHA256_128BITTAG,
    'HMAC_SHA256_256BITTAG':
        mac.mac_key_templates.HMAC_SHA256_256BITTAG,
    'HMAC_SHA512_256BITTAG':
        mac.mac_key_templates.HMAC_SHA512_256BITTAG,
    'HMAC_SHA512_512BITTAG':
        mac.mac_key_templates.HMAC_SHA512_512BITTAG,
    'ECDSA_P256':
        signature.signature_key_templates.ECDSA_P256,
    'ECDSA_P256_RAW':
        signature.signature_key_templates.ECDSA_P256_RAW,
    'ECDSA_P384':
        signature.signature_key_templates.ECDSA_P384,
    'ECDSA_P384_SHA384':
        signature.signature_key_templates.ECDSA_P384_SHA384,
    'ECDSA_P384_SHA512':
        signature.signature_key_templates.ECDSA_P384_SHA512,
    'ECDSA_P521':
        signature.signature_key_templates.ECDSA_P521,
    'ECDSA_P256_IEEE_P1363':
        signature.signature_key_templates.ECDSA_P256_IEEE_P1363,
    'ECDSA_P384_IEEE_P1363':
        signature.signature_key_templates.ECDSA_P384_IEEE_P1363,
    'ECDSA_P384_SHA384_IEEE_P1363':
        signature.signature_key_templates.ECDSA_P384_SHA384_IEEE_P1363,
    'ECDSA_P521_IEEE_P1363':
        signature.signature_key_templates.ECDSA_P521_IEEE_P1363,
    'ED25519':
        signature.signature_key_templates.ED25519,
    'RSA_SSA_PKCS1_3072_SHA256_F4':
        signature.signature_key_templates.RSA_SSA_PKCS1_3072_SHA256_F4,
    'RSA_SSA_PKCS1_4096_SHA512_F4':
        signature.signature_key_templates.RSA_SSA_PKCS1_4096_SHA512_F4,
    'RSA_SSA_PSS_3072_SHA256_SHA256_32_F4':
        signature.signature_key_templates.RSA_SSA_PSS_3072_SHA256_SHA256_32_F4,
    'RSA_SSA_PSS_4096_SHA512_SHA512_64_F4':
        signature.signature_key_templates.RSA_SSA_PSS_4096_SHA512_SHA512_64_F4,
    'AES_CMAC_PRF':
        prf.prf_key_templates.AES_CMAC,
    'HMAC_SHA256_PRF':
        prf.prf_key_templates.HMAC_SHA256,
    'HMAC_SHA512_PRF':
        prf.prf_key_templates.HMAC_SHA512,
    'HKDF_SHA256':
        prf.prf_key_templates.HKDF_SHA256,
    'JWT_HS256':
        jwt.jwt_hs256_template(),
    'JWT_HS256_RAW':
        jwt.raw_jwt_hs256_template(),
    'JWT_HS384':
        jwt.jwt_hs384_template(),
    'JWT_HS384_RAW':
        jwt.raw_jwt_hs384_template(),
    'JWT_HS512':
        jwt.jwt_hs512_template(),
    'JWT_HS512_RAW':
        jwt.raw_jwt_hs512_template(),
    'JWT_ES256':
        jwt.jwt_es256_template(),
    'JWT_ES256_RAW':
        jwt.raw_jwt_es256_template(),
    'JWT_ES384':
        jwt.jwt_es384_template(),
    'JWT_ES384_RAW':
        jwt.raw_jwt_es384_template(),
    'JWT_ES512':
        jwt.jwt_es512_template(),
    'JWT_ES512_RAW':
        jwt.raw_jwt_es512_template(),
    'JWT_RS256_2048_F4':
        jwt.jwt_rs256_2048_f4_template(),
    'JWT_RS256_2048_F4_RAW':
        jwt.raw_jwt_rs256_2048_f4_template(),
    'JWT_RS256_3072_F4':
        jwt.jwt_rs256_3072_f4_template(),
    'JWT_RS256_3072_F4_RAW':
        jwt.raw_jwt_rs256_3072_f4_template(),
    'JWT_RS384_3072_F4':
        jwt.jwt_rs384_3072_f4_template(),
    'JWT_RS384_3072_F4_RAW':
        jwt.raw_jwt_rs384_3072_f4_template(),
    'JWT_RS512_4096_F4':
        jwt.jwt_rs512_4096_f4_template(),
    'JWT_RS512_4096_F4_RAW':
        jwt.raw_jwt_rs512_4096_f4_template(),
    'JWT_PS256_2048_F4':
        jwt.jwt_ps256_2048_f4_template(),
    'JWT_PS256_2048_F4_RAW':
        jwt.raw_jwt_ps256_2048_f4_template(),
    'JWT_PS256_3072_F4':
        jwt.jwt_ps256_3072_f4_template(),
    'JWT_PS256_3072_F4_RAW':
        jwt.raw_jwt_ps256_3072_f4_template(),
    'JWT_PS384_3072_F4':
        jwt.jwt_ps384_3072_f4_template(),
    'JWT_PS384_3072_F4_RAW':
        jwt.raw_jwt_ps384_3072_f4_template(),
    'JWT_PS512_4096_F4':
        jwt.jwt_ps512_4096_f4_template(),
    'JWT_PS512_4096_F4_RAW':
        jwt.raw_jwt_ps512_4096_f4_template(),
}


class MetadataServicer(testing_api_pb2_grpc.MetadataServicer):
  """A service with metadata about the server."""

  def GetServerInfo(
      self, request: testing_api_pb2.ServerInfoRequest,
      context: grpc.ServicerContext) -> testing_api_pb2.ServerInfoResponse:
    """Returns information about the server."""
    return testing_api_pb2.ServerInfoResponse(language='python')


class KeysetServicer(testing_api_pb2_grpc.KeysetServicer):
  """A service for testing Keyset operations."""

  def GetTemplate(
      self, request: testing_api_pb2.KeysetTemplateRequest,
      context: grpc.ServicerContext) -> testing_api_pb2.KeysetTemplateResponse:
    """Returns the key template for the given template name."""
    if request.template_name not in _KEY_TEMPLATE:
      return testing_api_pb2.KeysetTemplateResponse(
          err='template %s not found' % request.template_name)
    return  testing_api_pb2.KeysetTemplateResponse(
        key_template=_KEY_TEMPLATE[request.template_name].SerializeToString())

  def Generate(
      self, request: testing_api_pb2.KeysetGenerateRequest,
      context: grpc.ServicerContext) -> testing_api_pb2.KeysetGenerateResponse:
    """Generates a keyset."""
    try:
      template = tink_pb2.KeyTemplate()
      template.ParseFromString(request.template)
      keyset_handle = tink.new_keyset_handle(template)
      keyset = io.BytesIO()
      cleartext_keyset_handle.write(
          tink.BinaryKeysetWriter(keyset), keyset_handle)
      return testing_api_pb2.KeysetGenerateResponse(keyset=keyset.getvalue())
    except tink.TinkError as e:
      return testing_api_pb2.KeysetGenerateResponse(err=str(e))

  def Public(
      self, request: testing_api_pb2.KeysetPublicRequest,
      context: grpc.ServicerContext) -> testing_api_pb2.KeysetPublicResponse:
    """Generates a public-key keyset from a private-key keyset."""
    try:
      private_keyset_handle = cleartext_keyset_handle.read(
          tink.BinaryKeysetReader(request.private_keyset))
      public_keyset_handle = private_keyset_handle.public_keyset_handle()
      public_keyset = io.BytesIO()
      cleartext_keyset_handle.write(
          tink.BinaryKeysetWriter(public_keyset), public_keyset_handle)
      return testing_api_pb2.KeysetPublicResponse(
          public_keyset=public_keyset.getvalue())
    except tink.TinkError as e:
      return testing_api_pb2.KeysetPublicResponse(err=str(e))

  def ToJson(
      self, request: testing_api_pb2.KeysetToJsonRequest,
      context: grpc.ServicerContext) -> testing_api_pb2.KeysetToJsonResponse:
    """Converts a keyset from binary to JSON format."""
    try:
      keyset_handle = cleartext_keyset_handle.read(
          tink.BinaryKeysetReader(request.keyset))
      json_keyset = io.StringIO()
      cleartext_keyset_handle.write(
          tink.JsonKeysetWriter(json_keyset), keyset_handle)
      return testing_api_pb2.KeysetToJsonResponse(
          json_keyset=json_keyset.getvalue())
    except tink.TinkError as e:
      return testing_api_pb2.KeysetToJsonResponse(err=str(e))

  def FromJson(
      self, request: testing_api_pb2.KeysetFromJsonRequest,
      context: grpc.ServicerContext) -> testing_api_pb2.KeysetFromJsonResponse:
    """Converts a keyset from JSON to binary format."""
    try:
      keyset_handle = cleartext_keyset_handle.read(
          tink.JsonKeysetReader(request.json_keyset))
      keyset = io.BytesIO()
      cleartext_keyset_handle.write(
          tink.BinaryKeysetWriter(keyset), keyset_handle)
      return testing_api_pb2.KeysetFromJsonResponse(keyset=keyset.getvalue())
    except tink.TinkError as e:
      return testing_api_pb2.KeysetFromJsonResponse(err=str(e))

  def ReadEncrypted(
      self, request: testing_api_pb2.KeysetReadEncryptedRequest,
      context: grpc.ServicerContext
  ) -> testing_api_pb2.KeysetReadEncryptedResponse:
    """Reads an encrypted keyset."""
    try:
      master_keyset_handle = cleartext_keyset_handle.read(
          tink.BinaryKeysetReader(request.master_keyset))
      master_aead = master_keyset_handle.primitive(aead.Aead)

      if request.keyset_reader_type == testing_api_pb2.KEYSET_READER_BINARY:
        reader = tink.BinaryKeysetReader(request.encrypted_keyset)
      elif request.keyset_reader_type == testing_api_pb2.KEYSET_READER_JSON:
        reader = tink.JsonKeysetReader(request.encrypted_keyset.decode('utf8'))
      else:
        raise ValueError('unknown keyset reader type')
      if request.HasField('associated_data'):
        keyset_handle = tink.read_keyset_handle_with_associated_data(
            reader, master_aead, request.associated_data.value)
      else:
        keyset_handle = tink.read_keyset_handle(reader, master_aead)

      keyset = io.BytesIO()
      cleartext_keyset_handle.write(
          tink.BinaryKeysetWriter(keyset), keyset_handle)
      return testing_api_pb2.KeysetReadEncryptedResponse(
          keyset=keyset.getvalue())
    except tink.TinkError as e:
      return testing_api_pb2.KeysetReadEncryptedResponse(err=str(e))

  def WriteEncrypted(
      self, request: testing_api_pb2.KeysetWriteEncryptedRequest,
      context: grpc.ServicerContext
  ) -> testing_api_pb2.KeysetWriteEncryptedResponse:
    """Writes an encrypted keyset."""
    try:
      master_keyset_handle = cleartext_keyset_handle.read(
          tink.BinaryKeysetReader(request.master_keyset))
      keyset_handle = cleartext_keyset_handle.read(
          tink.BinaryKeysetReader(request.keyset))
      master_aead = master_keyset_handle.primitive(aead.Aead)

      if request.keyset_writer_type == testing_api_pb2.KEYSET_WRITER_BINARY:
        encrypted_keyset = io.BytesIO()
        writer = tink.BinaryKeysetWriter(encrypted_keyset)
        if request.HasField('associated_data'):
          keyset_handle.write_with_associated_data(
              writer, master_aead, request.associated_data.value)
        else:
          keyset_handle.write(writer, master_aead)
        return testing_api_pb2.KeysetWriteEncryptedResponse(
            encrypted_keyset=encrypted_keyset.getvalue())
      elif request.keyset_writer_type == testing_api_pb2.KEYSET_WRITER_JSON:
        encrypted_keyset = io.StringIO()
        writer = tink.JsonKeysetWriter(encrypted_keyset)
        if request.HasField('associated_data'):
          keyset_handle.write_with_associated_data(
              writer, master_aead, request.associated_data.value)
        else:
          keyset_handle.write(writer, master_aead)
        return testing_api_pb2.KeysetWriteEncryptedResponse(
            encrypted_keyset=encrypted_keyset.getvalue().encode('utf8'))
      else:
        raise ValueError('unknown keyset writer type')
    except tink.TinkError as e:
      return testing_api_pb2.KeysetWriteEncryptedResponse(err=str(e))


class AeadServicer(testing_api_pb2_grpc.AeadServicer):
  """A service for testing AEAD encryption."""

  def Create(self, request: testing_api_pb2.CreationRequest,
             context: grpc.ServicerContext) -> testing_api_pb2.CreationResponse:
    """Creates an AEAD without using it."""
    try:
      keyset_handle = cleartext_keyset_handle.read(
          tink.BinaryKeysetReader(request.annotated_keyset.serialized_keyset))
      keyset_handle.primitive(aead.Aead)
      return testing_api_pb2.CreationResponse()
    except tink.TinkError as e:
      return testing_api_pb2.CreationResponse(err=str(e))

  def Encrypt(
      self, request: testing_api_pb2.AeadEncryptRequest,
      context: grpc.ServicerContext) -> testing_api_pb2.AeadEncryptResponse:
    """Encrypts a message."""
    keyset_handle = cleartext_keyset_handle.read(
        tink.BinaryKeysetReader(request.annotated_keyset.serialized_keyset))
    p = keyset_handle.primitive(aead.Aead)
    try:
      ciphertext = p.encrypt(request.plaintext, request.associated_data)
      return testing_api_pb2.AeadEncryptResponse(ciphertext=ciphertext)
    except tink.TinkError as e:
      return testing_api_pb2.AeadEncryptResponse(err=str(e))

  def Decrypt(
      self, request: testing_api_pb2.AeadDecryptRequest,
      context: grpc.ServicerContext) -> testing_api_pb2.AeadDecryptResponse:
    """Decrypts a message."""
    keyset_handle = cleartext_keyset_handle.read(
        tink.BinaryKeysetReader(request.annotated_keyset.serialized_keyset))
    p = keyset_handle.primitive(aead.Aead)
    try:
      plaintext = p.decrypt(request.ciphertext, request.associated_data)
      return testing_api_pb2.AeadDecryptResponse(plaintext=plaintext)
    except tink.TinkError as e:
      return testing_api_pb2.AeadDecryptResponse(err=str(e))


class StreamingAeadServicer(testing_api_pb2_grpc.StreamingAeadServicer):
  """A service for testing StreamingAEAD encryption."""

  def Create(self, request: testing_api_pb2.CreationRequest,
             context: grpc.ServicerContext) -> testing_api_pb2.CreationResponse:
    """Creates a Streaming Aead without using it."""
    try:
      keyset_handle = cleartext_keyset_handle.read(
          tink.BinaryKeysetReader(request.annotated_keyset.serialized_keyset))
      keyset_handle.primitive(streaming_aead.StreamingAead)
      return testing_api_pb2.CreationResponse()
    except tink.TinkError as e:
      return testing_api_pb2.CreationResponse(err=str(e))

  def Encrypt(
      self, request: testing_api_pb2.StreamingAeadEncryptRequest,
      context: grpc.ServicerContext
  ) -> testing_api_pb2.StreamingAeadEncryptResponse:
    """Encrypts a message."""
    try:
      keyset_handle = cleartext_keyset_handle.read(
          tink.BinaryKeysetReader(request.annotated_keyset.serialized_keyset))
      p = keyset_handle.primitive(streaming_aead.StreamingAead)
      ciphertext_destination = bytes_io.BytesIOWithValueAfterClose()
      with p.new_encrypting_stream(ciphertext_destination,
                                   request.associated_data) as plaintext_stream:
        plaintext_stream.write(request.plaintext)
      return testing_api_pb2.StreamingAeadEncryptResponse(
          ciphertext=ciphertext_destination.value_after_close())
    except tink.TinkError as e:
      return testing_api_pb2.StreamingAeadEncryptResponse(err=str(e))

  def Decrypt(
      self, request: testing_api_pb2.StreamingAeadDecryptRequest,
      context: grpc.ServicerContext
  ) -> testing_api_pb2.StreamingAeadDecryptResponse:
    """Decrypts a message."""
    try:
      keyset_handle = cleartext_keyset_handle.read(
          tink.BinaryKeysetReader(request.annotated_keyset.serialized_keyset))
      p = keyset_handle.primitive(streaming_aead.StreamingAead)
      stream = io.BytesIO(request.ciphertext)
      with p.new_decrypting_stream(stream, request.associated_data) as s:
        plaintext = s.read()
      return testing_api_pb2.StreamingAeadDecryptResponse(plaintext=plaintext)
    except tink.TinkError as e:
      return testing_api_pb2.StreamingAeadDecryptResponse(err=str(e))


class DeterministicAeadServicer(testing_api_pb2_grpc.DeterministicAeadServicer):
  """A service for testing Deterministic AEAD encryption."""

  def Create(self, request: testing_api_pb2.CreationRequest,
             context: grpc.ServicerContext) -> testing_api_pb2.CreationResponse:
    """Creates a Deterministic AEAD without using it."""
    try:
      keyset_handle = cleartext_keyset_handle.read(
          tink.BinaryKeysetReader(request.annotated_keyset.serialized_keyset))
      keyset_handle.primitive(daead.DeterministicAead)
      return testing_api_pb2.CreationResponse()
    except tink.TinkError as e:
      return testing_api_pb2.CreationResponse(err=str(e))

  def EncryptDeterministically(
      self, request: testing_api_pb2.DeterministicAeadEncryptRequest,
      context: grpc.ServicerContext
  ) -> testing_api_pb2.DeterministicAeadEncryptResponse:
    """Encrypts a message."""
    keyset_handle = cleartext_keyset_handle.read(
        tink.BinaryKeysetReader(request.annotated_keyset.serialized_keyset))
    p = keyset_handle.primitive(daead.DeterministicAead)
    try:
      ciphertext = p.encrypt_deterministically(request.plaintext,
                                               request.associated_data)
      return testing_api_pb2.DeterministicAeadEncryptResponse(
          ciphertext=ciphertext)
    except tink.TinkError as e:
      return testing_api_pb2.DeterministicAeadEncryptResponse(err=str(e))

  def DecryptDeterministically(
      self, request: testing_api_pb2.DeterministicAeadDecryptRequest,
      context: grpc.ServicerContext
  ) -> testing_api_pb2.DeterministicAeadDecryptResponse:
    """Decrypts a message."""
    keyset_handle = cleartext_keyset_handle.read(
        tink.BinaryKeysetReader(request.annotated_keyset.serialized_keyset))
    p = keyset_handle.primitive(daead.DeterministicAead)
    try:
      plaintext = p.decrypt_deterministically(request.ciphertext,
                                              request.associated_data)
      return testing_api_pb2.DeterministicAeadDecryptResponse(
          plaintext=plaintext)
    except tink.TinkError as e:
      return testing_api_pb2.DeterministicAeadDecryptResponse(err=str(e))


class MacServicer(testing_api_pb2_grpc.MacServicer):
  """A service for testing MACs."""

  def Create(self, request: testing_api_pb2.CreationRequest,
             context: grpc.ServicerContext) -> testing_api_pb2.CreationResponse:
    """Creates a MAC without using it."""
    try:
      keyset_handle = cleartext_keyset_handle.read(
          tink.BinaryKeysetReader(request.annotated_keyset.serialized_keyset))
      keyset_handle.primitive(mac.Mac)
      return testing_api_pb2.CreationResponse()
    except tink.TinkError as e:
      return testing_api_pb2.CreationResponse(err=str(e))

  def ComputeMac(
      self, request: testing_api_pb2.ComputeMacRequest,
      context: grpc.ServicerContext) -> testing_api_pb2.ComputeMacResponse:
    """Computes a MAC."""
    try:
      keyset_handle = cleartext_keyset_handle.read(
          tink.BinaryKeysetReader(request.annotated_keyset.serialized_keyset))
      p = keyset_handle.primitive(mac.Mac)
      mac_value = p.compute_mac(request.data)
      return testing_api_pb2.ComputeMacResponse(mac_value=mac_value)
    except tink.TinkError as e:
      return testing_api_pb2.ComputeMacResponse(err=str(e))

  def VerifyMac(
      self, request: testing_api_pb2.VerifyMacRequest,
      context: grpc.ServicerContext) -> testing_api_pb2.VerifyMacResponse:
    """Verifies a MAC value."""
    try:
      keyset_handle = cleartext_keyset_handle.read(
          tink.BinaryKeysetReader(request.annotated_keyset.serialized_keyset))
      p = keyset_handle.primitive(mac.Mac)
      p.verify_mac(request.mac_value, request.data)
      return testing_api_pb2.VerifyMacResponse()
    except tink.TinkError as e:
      return testing_api_pb2.VerifyMacResponse(err=str(e))


class HybridServicer(testing_api_pb2_grpc.HybridServicer):
  """A service for testing hybrid encryption and decryption."""

  def CreateHybridEncrypt(
      self, request: testing_api_pb2.CreationRequest,
      context: grpc.ServicerContext) -> testing_api_pb2.CreationResponse:
    """Creates a HybridEncrypt without using it."""
    try:
      keyset_handle = cleartext_keyset_handle.read(
          tink.BinaryKeysetReader(request.annotated_keyset.serialized_keyset))
      keyset_handle.primitive(hybrid.HybridEncrypt)
      return testing_api_pb2.CreationResponse()
    except tink.TinkError as e:
      return testing_api_pb2.CreationResponse(err=str(e))

  def CreateHybridDecrypt(
      self, request: testing_api_pb2.CreationRequest,
      context: grpc.ServicerContext) -> testing_api_pb2.CreationResponse:
    """Creates a HybridDecrypt without using it."""
    try:
      keyset_handle = cleartext_keyset_handle.read(
          tink.BinaryKeysetReader(request.annotated_keyset.serialized_keyset))
      keyset_handle.primitive(hybrid.HybridDecrypt)
      return testing_api_pb2.CreationResponse()
    except tink.TinkError as e:
      return testing_api_pb2.CreationResponse(err=str(e))

  def Encrypt(
      self, request: testing_api_pb2.HybridEncryptRequest,
      context: grpc.ServicerContext) -> testing_api_pb2.HybridEncryptResponse:
    """Encrypts a message."""
    try:
      public_keyset_handle = cleartext_keyset_handle.read(
          tink.BinaryKeysetReader(
              request.public_annotated_keyset.serialized_keyset))
      p = public_keyset_handle.primitive(hybrid.HybridEncrypt)
      ciphertext = p.encrypt(request.plaintext, request.context_info)
      return testing_api_pb2.HybridEncryptResponse(ciphertext=ciphertext)
    except tink.TinkError as e:
      return testing_api_pb2.HybridEncryptResponse(err=str(e))

  def Decrypt(
      self, request: testing_api_pb2.HybridDecryptRequest,
      context: grpc.ServicerContext) -> testing_api_pb2.HybridDecryptResponse:
    """Decrypts a message."""
    try:
      private_keyset_handle = cleartext_keyset_handle.read(
          tink.BinaryKeysetReader(
              request.private_annotated_keyset.serialized_keyset))
      p = private_keyset_handle.primitive(hybrid.HybridDecrypt)
      plaintext = p.decrypt(request.ciphertext, request.context_info)
      return testing_api_pb2.HybridDecryptResponse(plaintext=plaintext)
    except tink.TinkError as e:
      return testing_api_pb2.HybridDecryptResponse(err=str(e))


class SignatureServicer(testing_api_pb2_grpc.SignatureServicer):
  """A service for testing signatures."""

  def CreatePublicKeySign(
      self, request: testing_api_pb2.CreationRequest,
      context: grpc.ServicerContext) -> testing_api_pb2.CreationResponse:
    """Creates a PublicKeySign without using it."""
    try:
      keyset_handle = cleartext_keyset_handle.read(
          tink.BinaryKeysetReader(request.annotated_keyset.serialized_keyset))
      keyset_handle.primitive(signature.PublicKeySign)
      return testing_api_pb2.CreationResponse()
    except tink.TinkError as e:
      return testing_api_pb2.CreationResponse(err=str(e))

  def CreatePublicKeyVerify(
      self, request: testing_api_pb2.CreationRequest,
      context: grpc.ServicerContext) -> testing_api_pb2.CreationResponse:
    """Creates a PublicKeyVerify without using it."""
    try:
      keyset_handle = cleartext_keyset_handle.read(
          tink.BinaryKeysetReader(request.annotated_keyset.serialized_keyset))
      keyset_handle.primitive(signature.PublicKeyVerify)
      return testing_api_pb2.CreationResponse()
    except tink.TinkError as e:
      return testing_api_pb2.CreationResponse(err=str(e))

  def Sign(
      self, request: testing_api_pb2.SignatureSignRequest,
      context: grpc.ServicerContext) -> testing_api_pb2.SignatureSignResponse:
    """Signs a message."""
    try:
      private_keyset_handle = cleartext_keyset_handle.read(
          tink.BinaryKeysetReader(
              request.private_annotated_keyset.serialized_keyset))
      p = private_keyset_handle.primitive(signature.PublicKeySign)
      signature_value = p.sign(request.data)
      return testing_api_pb2.SignatureSignResponse(signature=signature_value)
    except tink.TinkError as e:
      return testing_api_pb2.SignatureSignResponse(err=str(e))

  def Verify(
      self, request: testing_api_pb2.SignatureVerifyRequest,
      context: grpc.ServicerContext) -> testing_api_pb2.SignatureVerifyResponse:
    """Verifies a signature."""
    try:
      public_keyset_handle = cleartext_keyset_handle.read(
          tink.BinaryKeysetReader(
              request.public_annotated_keyset.serialized_keyset))
      p = public_keyset_handle.primitive(signature.PublicKeyVerify)
      p.verify(request.signature, request.data)
      return testing_api_pb2.SignatureVerifyResponse()
    except tink.TinkError as e:
      return testing_api_pb2.SignatureVerifyResponse(err=str(e))


class PrfSetServicer(testing_api_pb2_grpc.PrfSetServicer):
  """A service for testing PrfSet."""

  def Create(self, request: testing_api_pb2.CreationRequest,
             context: grpc.ServicerContext) -> testing_api_pb2.CreationResponse:
    """Creates a PrfSet without using it."""
    try:
      keyset_handle = cleartext_keyset_handle.read(
          tink.BinaryKeysetReader(request.annotated_keyset.serialized_keyset))
      keyset_handle.primitive(prf.PrfSet)
      return testing_api_pb2.CreationResponse()
    except tink.TinkError as e:
      return testing_api_pb2.CreationResponse(err=str(e))

  def KeyIds(
      self, request: testing_api_pb2.PrfSetKeyIdsRequest,
      context: grpc.ServicerContext) -> testing_api_pb2.PrfSetKeyIdsResponse:
    """Returns all key IDs and the primary key ID."""
    try:
      keyset_handle = cleartext_keyset_handle.read(
          tink.BinaryKeysetReader(request.annotated_keyset.serialized_keyset))
      p = keyset_handle.primitive(prf.PrfSet)
      prfs = p.all()
      response = testing_api_pb2.PrfSetKeyIdsResponse()
      response.output.primary_key_id = p.primary_id()
      response.output.key_id.extend(prfs.keys())
      return response
    except tink.TinkError as e:
      return testing_api_pb2.PrfSetKeyIdsResponse(err=str(e))

  def Compute(
      self, request: testing_api_pb2.PrfSetComputeRequest,
      context: grpc.ServicerContext) -> testing_api_pb2.PrfSetComputeResponse:
    """Computes the output of one PRF."""
    try:
      keyset_handle = cleartext_keyset_handle.read(
          tink.BinaryKeysetReader(request.annotated_keyset.serialized_keyset))
      f = keyset_handle.primitive(prf.PrfSet).all()[request.key_id]
      return testing_api_pb2.PrfSetComputeResponse(
          output=f.compute(request.input_data, request.output_length))
    except tink.TinkError as e:
      return testing_api_pb2.PrfSetComputeResponse(err=str(e))