incountry 4.1.0

InCountry Storage SDK

InCountry Storage SDK

Table of contents

• Installation
• Countries list
• Quickstart guide
• Storage configuration
  • Extra Storage options
  • Request Options
  • oAuth options configuration
  • Encryption key/secret
• Usage
  • Writing data to storage
    • List of available record fields
    • Date fields
  • Batches
  • Reading stored data
  • Find records
    • Sorting
    • Find error handling
  • Find one record
  • Delete records
• Attaching files to a record
  • Adding attachments
  • Deleting attachments
  • Downloading attachments
  • Working with attachment meta info
• Health check
• Data migration and key rotation support
• AWS KMS integration
• Error handling
• Custom encryption support
• Testing Locally

Installation

To install Python SDK, use pipenv (or pip), as follows:

$ pipenv install incountry

Countries List

To get the full list of supported countries and their codes, please follow this link.

Quickstart guide

To access your data in InCountry Platform by using Python SDK, you need to create an instance of the Storage class. You can retrieve the client_id, client_secret and environment_id variables from your dashboard on InCountry Portal.

from incountry import Storage, SecretKeyAccessor

storage = Storage(
    client_id="<client_id>",
    client_secret="<client_secret>",
    environment_id="<environment_id>",
    secret_key_accessor=SecretKeyAccessor(lambda: "<encryption_secret>"),
)

Storage Configuration

Below you can find a full list of possible configuration options for creating a Storage instance.

class Storage:
    def __init__(
        self,
        # Required.
        # Can be set via INC_ENVIRONMENT_ID env variable.
        environment_id: Optional[str] = None,

        # Required when using oAuth authorization.
        # Can be set via INC_CLIENT_ID env variable.
        client_id: Optional[str] = None,

        # Required when using oAuth authorization.
        # Can be set via INC_CLIENT_SECRET env variable.
        client_secret: Optional[str] = None,

        # Used when OAuth token is already acquired prior to Storage initialization.
        # Mutually exclusive with client_id, client_secret.
        oauth_token: Optional[str] = None,

        # Instance of the SecretKeyAccessor class.
        # Used to fetch the encryption secret.
        secret_key_accessor: Optional[SecretKeyAccessor] = None,

        # Optional. Defines API URL.
        # Can be set via INC_ENDPOINT env variable.
        endpoint: Optional[str] = None,

        # Optional. If False, encryption is not applied.
        encrypt: Optional[bool] = True,

        # Optional. If True enables the additional debug logging.
        debug: Optional[bool] = False,

         # Optional. Used to fine-tune some configurations.
        options: Optional[Dict[str, Any]] = {},

        # Optional. List of custom encryption configurations.
        custom_encryption_configs: Optional[List[dict]] = None,
    ):
        ...

---

Extra Storage options

You can use the options parameter to tweak some SDK configurations:

storage = Storage(
    ...,
    options={
        "http_options": {
            "timeout": int,         # Defines http requests timeout, in seconds.
                                    # Default value is 30
            "retryBaseDelay": int,  # Sets a custom initial http request retry delay in seconds.
                                    # Retries when receiving HTTP 429 code from the server.
                                    # Default value is 1
            "retryMaxDelay": int,   # Sets a custom maximum retry delay in seconds.
                                    # Default value is 32
        },

        "auth_endpoints": dict,     # A custom endpoints map that should be used for
                                    # fetching oAuth tokens

        "countries_endpoint": str,  # If your PoPAPI configuration relies on a custom
                                    # PoPAPI server (rather than the default one)
                                    # use the `countries_endpoint` option to specify
                                    # the endpoint responsible for fetching the list
                                    # of supported countries

        "endpoint_mask": str,       # Defines API base hostname part to use.
                                    # If set, all requests will be sent to
                                    # https://${country}${endpoint_mask} host
                                    # instead of the default one
                                    # (https://${country}-mt-01.api.incountry.io)

        "normalize_keys": bool,     # If set to True all the string record keys will be stored in lowercase
                                    # (except 'body' and 'precommit_body').

        "hash_search_keys": bool,   # If set to False key1-key20 will be stored unhashed, enabling
                                    # partial match search using 'search_keys'
    }
)

Request Options

Every Storage method allows to pass custom HTTP headers via request_options parameter, e.g.:

storage.write(
    country="us",
    record_key="user_1",
    ...,
    request_options={
        "http_headers": {
            "request-id": "<uuid>",
        },
    },
)

oAuth options configuration

The SDK allows to precisely configure oAuth authorization endpoints (if needed). Use this option only if your plan configuration requires so.

Below you can find the example of how to create a storage instance with custom oAuth endpoints:

storage = Storage(
    client_id="<client_id>",
    client_secret="<client_secret>",
    environment_id="<environment_id>",
    secret_key_accessor=SecretKeyAccessor(lambda: "<encryption_secret>"),
    options={
        "auth_endpoints": {
            "default": "<default_auth_endpoint>",
            "emea": "<auth_endpoint_for_emea_region>",
            "apac": "<auth_endpoint_for_apac_region>",
            "amer": "<auth_endpoint_for_amer_region>",
        },
    },
)

Encryption key/secret

The secret_key_accessor variable is used to pass a key or secret used for data encryption.

The SecretKeyAccessor class constructor allows you to pass a function that should return either a string representing your secret or a dictionary (we call it secrets_data object):

{
  "secrets": [{
       "secret": str,
       "version": int, # Should be an integer greater than or equal to 0
       "isKey": bool,  # Should be True only for user-defined encryption keys
    }
  }, ....],
  "currentVersion": int,
}

Note: even though SDK uses PBKDF2 to generate a cryptographically strong encryption key, you must ensure that you provide a secret/password which follows the modern security best practices and standards.

The secrets_data variable allows you to specify multiple keys/secrets which SDK will use for data decryption based on the version of the key or secret used for encryption. Meanwhile SDK will encrypt data only by using a key (or secret) which matches the currentVersion parameter provided in the secrets_data object.

This enables the flexibility required to support Key Rotation policies when secrets (or keys) must be changed with time. The SDK will encrypt data by using the current secret (or key) while maintaining the ability to decrypt data records that were encrypted with old secrets (or keys). The SDK also provides a method for data migration which allows you to re-encrypt data with the newest secret (or key). For details please see the migrate method.

The SDK allows you to use custom encryption keys, instead of secrets. Please note that a user-defined encryption key should be a base64-encoded 32-bytes-long key as required by AES-256 cryptographic algorithm.

Below you can find several examples of how you can use SecretKeyAccessor module.

# Get a secret from a variable
from incountry import SecretKeyAccessor

password = "password"
secret_key_accessor = SecretKeyAccessor(lambda: password)

# Get secrets via an HTTP request
from incountry import SecretKeyAccessor
import requests as req

def get_secrets_data():
    url = "<your_secret_url>"
    r = req.get(url)
    return r.json() # assuming response is a `secrets_data` object

secret_key_accessor = SecretKeyAccessor(get_secrets_data)

Usage

Writing data to Storage

Use the write method in order to create/replace a record (by record_key).

def write(self, country: str, record_key: str, request_options: dict = {}, **record_data: Union[str, int]) -> Dict[str, TRecord]:
    ...


# write returns created record dict on success
{
    "record": Dict
}

Below you can find an example of how you can use the write method.

write_result = storage.write(
    country="us",
    record_key="user_1",
    body="some PII data",
    profile_key="customer",
    range_key1=10000,
    key1="english",
    key2="rolls-royce",
)

# write_result would be as follows
write_result = {
    "record": {
        "record_key": "user_1",
        "body": "some PII data",
        "profile_key": "customer",
        "range_key1": 10000,
        "key1": "english",
        "key2": "rolls-royce",
    }
}

For the list of possible record_data kwargs, see the section below.

List of available record fields

v3.0.0 release introduced a series of new fields available for data storage. Below you can find the full list of all the fields available for storage in InCountry Platform along with their types and storage methods. Each field is either encrypted, hashed or stored as is:

# String fields, hashed
record_key
key1
key2
key3
key4
key5
key6
key7
key8
key9
key10
key11
key12
key13
key14
key15
key16
key17
key18
key19
key20
profile_key
service_key1
service_key2
service_key3
service_key4
service_key5
parent_key

# String fields, encrypted
body
precommit_body

# Date fields, plain
expires_at

# Int fields, plain
range_key1
range_key2
range_key3
range_key4
range_key5
range_key6
range_key7
range_key8
range_key9
range_key10

Date fields

Use the created_at and updated_at fields to access date-related information about records. The created_at field stores a date when a record was initially created in the target country. The updated_at field stores a date of the latest write operation for the given record_key.

Use expires_at field to limit the lifespan of a record stored in InCountry Platform. The record will be automatically deleted upon reaching the specified date.

Below is an example of how to create a record that will be deleted in a day:

from datetime import datetime, timedelta, timezone
...

deletion_date = datetime.now(timezone.utc) + timedelta(days=1)

write_result = storage.write(
    country="us",
    record_key="user_1",
    expires_at=deletion_date,
)

# write_result would be as follows
write_result = {
    "record": {
        "record_key": "user_1",
        "expires_at": datetime.datetime(2021, 8, 11, 12, 36, 52, 607534, tzinfo=datetime.timezone.utc),
        "created_at": datetime.datetime(2021, 7, 26, 15, 34, 28, 910266, tzinfo=datetime.timezone.utc),
        "updated_at": datetime.datetime(2021, 8, 10, 12, 36, 54, 879828, tzinfo=datetime.timezone.utc),
    }
}

---

NOTE

All dates are stored in UTC timezone and converted to UTC timezone in case of expires_at.

It's highly recommended to provide timezone-aware datetime objects to avoid any timezone-related issues in future.

E.g., to create an object representing the current time in UTC you can call datetime.now(timezone.utc)

Please avoid using datetime.utcnow() as it returns naive datetime object without timezone information, which may result in wrong timezone conversion when passed as expires_at value.

---

Batches

Use the batch_write method to create/replace multiple records at once.

def batch_write(self, country: str, records: List[TRecord], request_options: dict = {}) -> Dict[str, List[TRecord]]:
    ...


# batch_write returns the following dict of created records
{
    "records": List
}

Below you can find an example of how to use this method.

batch_result = storage.batch_write(
    country="us",
    records=[
        {"record_key": "key1", "body": "body1", ...},
        {"record_key": "key2", "body": "body2", ...},
    ],
)

# batch_result would be as follows
batch_result = {
    "records": [
        {"record_key": "key1", "body": "body1", ...},
        {"record_key": "key2", "body": "body2", ...},
    ]
}

Reading stored data

You can read the stored data records by its record_key using the read method.

def read(self, country: str, record_key: str, request_options: dict = {}) -> Dict[str, TRecord]:
    ...


# The read method returns the record dictionary if the record is found
{
    "record": Dict
}

The read method usage is as follows:

read_result = storage.read(country="us", record_key="user1")

# read_result would be as follows
read_result = {
    "record": {
        "record_key": "user_1",
        "body": "some PII data",
        "profile_key": "customer",
        "range_key1": 10000,
        "key1": "english",
        "key2": "rolls-royce",
        "created_at": datetime.datetime(...),
        "updated_at": datetime.datetime(...),
    }
}

Find records

You can look up for records by keys or version using the find method.

def find(
        self,
        country: str,
        limit: Optional[int] = FIND_LIMIT,
        offset: Optional[int] = 0,
        sort: Optional[List[TSortFilter]] = None,
        request_options: dict = {},
        **filters: Union[TIntFilter, TStringFilter],
    ) -> Dict[str, Any]:
    ...

Note: SDK returns 100 records at most.

The returned object looks like the following:

{
    "data": List,
    "errors": List, # optional
    "meta": {
        "limit": int,
        "offset": int,
        "total": int,  # total records matching filter, ignoring limit
    }
}

You can use the following options to look up for records by hashed string keys from the list above:

# single value
key1="value1" # records with key1 equal to "value1"

# list of values
key2=["value1", "value2"] # records with key2 equal to "value1" or "value2"

# dict with $not operator
key3={"$not": "value1"} # records with key3 not equal "value1"
key4={"$not": ["value1", "value2"]} # records with key4 equal to neither "value1" or "value2"

You can use special search_keys filter to search records by partial match (similar to LIKE SQL operator) among record's text fields key1, ..., key20.

search_keys="text to find"

---

NOTE

search_keys cannot be used in combination with any of key1, key2, ..., key20 keys and works only in combination with non-hashing Storage mode (with hash_search_keys Storage option set to False).

---

You can use the following options to look up for records by int keys from the list above:

# single value
range_key1=1 # records with range_key1 equal to 1

# list of values
range_key2=[1, 2] # records with range_key2 equal to 1 or 2

# dictionary with comparison operators
range_key3={"$gt": 1} # records with range_key3 greater than 1
range_key4={"$gte": 1} # records with range_key4 greater than or equal to 1
range_key5={"$lt": 1} # records with range_key5 less than 1
range_key6={"$lte": 1} # records with range_key6 less than or equal to 1

# you can combine different comparison operators
range_key7={"$gt": 1, "$lte": 10} # records with range_key7 greater than 1 and less than or equal to 10

# you cannot combine similar comparison operators - e.g. $gt and $gte, $lt and $lte

You can use the following option to look up for records by version (encryption key version):

# single value
version=1 # records with version equal to 1

# list of values
version=[1, 2] # records with version equal to 1 or 2

# dictionary with $not operator
version={"$not": 1} # records with version not equal 1
version={"$not": [1, 2]} # records with version equal to neither 1 or 2

Sorting

You can use sort parameter to sort the returned records by one or multiple keys. SDK currently supports sorting by key1...key20, range_key1...range_key10, created_at, updated_at, expires_at keys.

from incountry import SortOrder

# you can use SordOrder enum class to specify sorting order
sort=[{"key1": SortOrder.ASC}, {"range_key1": SortOrder.DESC}, {"created_at": SortOrder.DESC}]

# or you can use string values
sort=[{"key1": "asc"}, {"range_key1": "desc"}, {"updated_at": "asc"}]

---

WARNING

To use sort in find() call for string keys key1...key20 you need to set hash_search_keys option to False.

---

Below you can find an example of how you can use the find method:

find_result = storage.find(country="us", limit=10, offset=10, key1="value1", key2=["value2", "value3"], sort=[{"key1": "asc"}])

# find_result would be as follows
find_result = {
    "data": [
        {
            "record_key": "<record_key>",
            "body": "<body>",
            "key1": "value1",
            "key2": "value2",
            "created_at": datetime.datetime(...),
            "updated_at": datetime.datetime(...),
            ...
        }
    ],
    "meta": {
        "limit": 10,
        "offset": 10,
        "total": 100,
    }
}

Find error handling

There may be a situation when the find method receives records that cannot be decrypted. For example, this may happen once the encryption key has been changed while the found data was encrypted with the older version of that key. In such cases data returned by the find() method will be as follows:

{
    "data": [...],  # successfully decrypted records
    "errors": [{
        "rawData",  # raw record which caused decryption error
        "error",    # decryption error description
    }, ...],
    "meta": { ... }
}

Find one record

If you need to find only one of the records matching a specific filter, you can use the find_one method.

def find_one(
        self,
        country: str,
        offset: Optional[int] = 0,
        sort: Optional[List[TSortFilter]] = None,
        request_options: dict = {},
        **filters: Union[TIntFilter, TStringFilter],
    ) -> Union[None, Dict[str, Dict]]:
    ...


# If a record is not found, the find_one method returns `None`. Otherwise it returns a record dictionary.
{
    "record": Dict
}

Below you can find the example of how to use the find_one method:

find_one_result = storage.find_one(country="us", key1="english", key2=["rolls-royce", "bmw"])

# find_one_result would be as follows
find_one_result = {
    "record": {
        "record_key": "user_1",
        "body": "some PII data",
        "profile_key": "customer",
        "range_key1": 10000,
        "key1": "english",
        "key2": "rolls-royce",
    }
}

Delete records

Use the delete method in order to delete a record from InCountry Platform. It is only possible by using the record_key field.

def delete(self, country: str, record_key: str, request_options: dict = {}) -> Dict[str, bool]:
    ...


# the delete method returns the following dictionary upon success
{
    "success": True
}

Below you can find the example of how to use the delete method:

delete_result = storage.delete(country="us", record_key="<record_key>")

# delete_result would be as follows
delete_result = {
    "success": True
}

Attaching files to a record

NOTE

Attachments are currently available for InCountry dedicated instances only. Please check your subscription plan for details. This may require specifying your dedicated instance endpoint when configuring Python SDK Storage.

---

InCountry Storage allows you to attach files to the previously created records. Attachments' meta information is available through the attachments field of record dictionary, returned by read, find and find_one methods.

read_result = storage.read(country="us", record_key="<record_key>")

# read_result with attachments would be as follows
read_result = {
    "record": {
        "record_key": "user_1",
        ...,
        "attachments": [
            {
                "created_at": datetime,
                "updated_at": datetime,
                "download_link": "<download_url>",
                "file_id": "<file_id>",
                "filename": "<file_name>.<ext>",
                "hash": "<hash>",
                "mime_type": "<mime_type>",
                "size": 1,
            },
        ],
    },
}

Adding attachments

---

Note:

The add_attachment method allows you to add or replace attachments. File data can be provided either as BinaryIO object or string with a path to the file in the file system.

def add_attachment(
    self,
    country: str,
    record_key: str,
    file: Union[BinaryIO, str],
    mime_type: str = None,
    upsert: bool = False,
    request_options: dict = {},
) -> Dict[str, Union[str, int, datetime]]:
    ...

# The add_attachment returns attachment_meta dictionary
{
    "attachment_meta": {
        "created_at": datetime,
        "updated_at": datetime,
        "download_link": "<download_url>",
        "file_id": "<file_id>",
        "filename": "<file_name>.<ext>",
        "hash": "<hash>",
        "mime_type": "<mime_type>",
        "size": 1,
    }
}

Example of usage:

# using file path
storage.add_attachment(country="us", record_key="<record_key>", file="./README.md")

# using BinaryIO object
with open("./README.md", "rb") as attachment_file:
    storage.add_attachment(country="<country>", record_key="<record_key>", file=attachment_file)

Deleting attachments

The delete_attachment method allows you to delete attachment using its file_id.

def delete_attachment(
    self, country: str, record_key: str, file_id: str, request_options: dict = {}
) -> Dict[str, bool]::
    ...

# the delete_attachment method returns the following dictionary upon success
{
    "success": True
}

Example of usage:

delete_result = storage.delete_attachment(country="us", record_key="<record_key>", file_id="<file_id>")

# delete_result would be as follows
delete_result = {
    "success": True
}

Downloading attachments

The get_attachment_file method allows you to download attachment contents. It returns dictionary with readable file body and filename.

def get_attachment_file(
    self, country: str, record_key: str, file_id: str, request_options: dict = {}
) -> Dict[str, Dict]:
    ...

# the get_attachment_file method returns the following dictionary upon success
{
    "attachment_data": {
        "filename": str,
        "file": BytesIO,
    }
}

Example of usage:

get_attachment_res = storage.get_attachment_file(country="us", record_key="<record_key>", file_id="<file_id>")
attachment_data = get_attachment_res["attachment_data"]

with open(attachment_data["filename"], "wb") as f:
    f.write(attachment_data["file"].read())

Working with attachment meta info

The get_attachment_meta method allows you to retrieve attachment's metadata using its file_id.

def get_attachment_meta(
    self, country: str, record_key: str, file_id: str, request_options: dict = {}
) -> Dict[str, Union[str, int, datetime]]:
    ...

# The get_attachment_meta returns attachment_meta dictionary
{
    "attachment_meta": {
        "created_at": datetime,
        "updated_at": datetime,
        "download_link": "<download_url>",
        "file_id": "<file_id>",
        "filename": "<file_name>.<ext>",
        "hash": "<hash>",
        "mime_type": "<mime_type>",
        "size": 1,
    }
}

Example of usage:

get_attachment_meta_res = storage.get_attachment_meta(country="us", record_key="<record_key>", file_id="<file_id>")

The update_attachment_meta method allows you to update attachment's metadata (MIME type and file name).

def update_attachment_meta(
    self,
    country: str,
    record_key: str,
    file_id: str,
    filename: str = None,
    mime_type: str = None,
    request_options: dict = {},
) -> Dict[str, Union[str, int, datetime]]:
    ...


# The update_attachment_meta returns attachment_meta dictionary
{
    "attachment_meta": {
        "created_at": datetime,
        "updated_at": datetime,
        "download_link": "<download_url>",
        "file_id": "<file_id>",
        "filename": "<file_name>.<ext>",
        "hash": "<hash>",
        "mime_type": "<mime_type>",
        "size": 1,
    }
}

Example of usage:

storage.update_attachment_meta(country="us", record_key="<record_key>", file_id="<file_id>", filename="new_file_name.txt", mime_type="text/plain")

Health check

The health_check method of Storage class allows you to check availability of a remote storage service by country identifier

def health_check(
    self,
    country: str,
    request_options: dict = {},
) -> Dict[str, bool]:

Example of usage:

response = storage.health_check(country="us")

# response would be as follows
response = {
    "result": True
}

Data Migration and Key Rotation support

Using secret_key_accessor which provides the secrets_data object enables key rotation and data migration support.

SDK introduces the migrate method which allows you to re-encrypt data encrypted with older versions of the secret.

def migrate(self, country: str, limit: Optional[int] = FIND_LIMIT, request_options: dict = {}) -> Dict[str, int]:
    ...


# The migrate method returns the following dictionary with meta information
{
    "migrated": int   # the number of records migrated
	"total_left": int # the number of records left to migrate (number of records with version
                      # different from `currentVersion` provided by `secret_key_accessor`)
}

You should specify the country parameter which you want to perform migration in. Additionally, you need to specify the limit parameter for a precise number of records to migrate.

Note: the maximum number of records that can be migrated per one request is 100.

For a detailed example of a migration script, please see /examples/full_migration.py

AWS KMS integration

InCountry Python SDK supports usage of any 32-byte (256-bit) AES key, including ones produced by AWS KMS symmetric master key (CMK).

The suggested use case assumes that AWS user already got his KMS encrypted data key (AES_256) generated. Afterwards the key gets decrypted using AWS Python client library (boto3) and then provided to InCountry Python SDK's SecretKeyAccessor.

For a detailed example of AWS KMS keys usage please see examples/aws-kms.py

Error Handling

InCountry Python SDK may throw the following Exceptions:

• StorageConfigValidationException - used for Storage options validation exceptions. Can be thrown when trying to create a Storage class instance.

• SecretsProviderException - can be thrown in runtime after Storage instance creation when an exception occurs trying to retrieve secrets.

• SecretsValidationException - can be thrown if secrets returned by SecretKeyAccessor instance are in the wrong format.

• InputValidationException - used for input validation exceptions. Can be thrown by all Storage public methods.

• StorageClientException - used for various exceptions related to Storage misconfiguration or wrong input data provision. All of the exception classes StorageConfigValidationException, SecretsProviderException, SecretsValidationException, InputValidationException are inherited from StorageClientException.

• StorageServerException - thrown if SDK fails to communicate with InCountry servers or if a server response validation fails.

• StorageServerResponseValidationException - thrown if a server response validation fails. Inherited from StorageServerException.

• StorageNetworkException - thrown if SDK failed to communicate with InCountry servers due to network issues, such as request timeout, unreachable endpoint etc. Inherited from StorageServerException.

• StorageAuthenticationException - can be thrown if SDK failed to authenticate in InCountry system with the provided credentials. Inherited from StorageServerException.

• StorageCryptoException - thrown during encryption/decryption procedures (both default and custom). This may be an indication of malformed/corrupted data or a wrong encryption key provided to the SDK.

• StorageException - general exception. Inherited by all the other exceptions.

We strongly recommend you to gracefully handle all the possible exceptions:

try:
    # use InCountry Storage instance here
except StorageConfigValidationException as e:
    # storage configuration error
except SecretsProviderException as e:
    # error trying to acquire secrets
except SecretsValidationException as e:
    # secrets validation error
except InputValidationException as e:
    # input validation error
except StorageClientException as e:
    # general error related to user somehow provided wrong input data
    # when creating storage or calling a storage method
except StorageServerResponseValidationException as e:
    # error when trying to validate server response
except StorageNetworkException as e:
    # network-related error
except StorageAuthenticationException as e:
    # authentication error
except StorageServerException as e:
    # general error during SDK<->server communication
except StorageCryptoException as e:
    # some encryption error
except StorageException as e:
    # general error
except Exception as e:
    # something else happened not related to InCountry SDK

Custom Encryption Support

SDK supports a capability to provide custom encryption/decryption methods if you decide to use your own algorithm instead of the default one.

The Storage constructor allows you to pass custom_encryption_configs parameters as an array of custom encryption configurations within the following schema which enables custom encryption:

{
    "encrypt": Callable,
    "decrypt": Callable,
    "isCurrent": bool,
    "version": str
}

Both encrypt and decrypt attributes should be functions implementing the following interface (with exactly the same argument names)

encrypt(input:str, key:bytes, key_version:int) -> str:
    ...

decrypt(input:str, key:bytes, key_version:int) -> str:
    ...

They should accept raw data to encrypt/decrypt, key data (represented as bytes array) and key version received from SecretKeyAccessor. The resulted encrypted/decrypted data should be a string.

---

NOTE

You should provide a specific encryption key through secrets_data passed to SecretKeyAccessor. This secret should use the isForCustomEncryption flag instead of the regular isKey flag.

secrets_data = {
  "secrets": [{
       "secret": "<secret for custom encryption>",
       "version": 1,
       "isForCustomEncryption": True,
    }
  }],
  "currentVersion": 1,
}

secret_accessor = SecretKeyAccessor(lambda: secrets_data)

---

The version attribute is used to differentiate custom encryptions from each other and from the default encryption as well. This way SDK will be able to successfully decrypt any old data if encryption changes with time.

The isCurrent attribute allows you to specify one of the custom encryption configurations which should be used for encryption. Only one encryption configuration can be set as "isCurrent": True.

If none of the encryption configurations have "isCurrent": True then the SDK will use default encryption configuration to encrypt stored data. At the same time it will preserve the ability to decrypt old data which was encrypted with custom encryption configuration (if any).

Below you can find the example of how you can set up the SDK to use custom encryption (using Fernet encryption method from https://cryptography.io/en/latest/fernet/)

import os

from incountry import InCrypto, SecretKeyAccessor, Storage
from cryptography.fernet import Fernet

def enc(input: str, key: bytes, key_version: int):
    # Fernet cipher from cryptography package accepts keys as base64-encoded string
    # while InCountry SDK decodes keys as bytes
    # thus we need to do some bytes-to-b64 encoding
    cipher = Fernet(InCrypto.b_to_base64(key))
    return cipher.encrypt(input.encode("utf8")).decode("utf8")

def dec(input: str, key: bytes, key_version: int):
    cipher = Fernet(InCrypto.b_to_base64(key))
    return cipher.decrypt(input.encode("utf8")).decode("utf8")

custom_encryption_configs = [
    {
        "encrypt": enc,
        "decrypt": dec,
        "version": "test",
        "isCurrent": True,
    }
]

key = os.urandom(InCrypto.KEY_LENGTH)  # Fernet uses 32-byte length keys

secret_key_accessor = SecretKeyAccessor(
    lambda: {
        "currentVersion": 1,
        "secrets": [{"secret": key, "version": 1, "isForCustomEncryption": True}],
    }
)

storage = Storage(
    environment_id="<env_id>",
    client_id="<client_id>",
    client_secret="<client_secret>",
    secret_key_accessor=secret_key_accessor,
    custom_encryption_configs=custom_encryption_configs,
)

storage.write(country="us", record_key="<record_key>", body="<body>")

Testing Locally

1. In terminal run pipenv run tests for unit tests
2. In terminal run pipenv run integrations to run integration tests