Welcome to VaultAPI’s documentation!

Contents:

VaultAPI - Main

vaultapi.main.enable_cors() → None

Enables CORS policy.

vaultapi.main.start(**kwargs) → None

Starter function for the API, which uses uvicorn server as trigger.

Keyword Arguments:

  • env_file – Env filepath to load the environment variables.

  • apikey – API Key to authenticate the server.

  • secret – Secret access key to access the secret content.

  • host – Hostname for the API server.

  • port – Port number for the API server.

  • workers – Number of workers for the uvicorn server.

  • database – FilePath to store the auth database that handles the authentication errors.

  • rate_limit – List of dictionaries with max_requests and seconds to apply as rate limit.

  • log_config – Logging configuration as a dict or a FilePath. Supports .yaml/.yml, .json or .ini formats.

Authenticator

async vaultapi.auth.validate(request: Request, apikey: HTTPAuthorizationCredentials) → None

Validates the auth request using HTTPBearer.

Parameters:

  • request – Takes the authorization header token as an argument.

  • apikey – Basic APIKey required for all the routes.

Raises:

  • APIResponse –

  • - 401 – If authorization is invalid.

  • - 403 – If host address is forbidden.

Database

vaultapi.database.table_exists(table_name: str) → bool

Function to check if a table exists in the database.

Parameters:

table_name – Name of the table to check.

vaultapi.database.create_table(table_name: str, columns: Union[List[str], Tuple[str]]) → None

Creates the table with the required columns.

Parameters:

  • table_name – Name of the table that has to be created.

  • columns – List of columns that has to be created.

vaultapi.database.get_secret(key: str, table_name: str) → str | None

Function to retrieve secret from database.

Parameters:

  • key – Name of the secret to retrieve.

  • table_name – Name of the table where the secret is stored.

Returns:

Returns the secret value.

Return type:

str

vaultapi.database.get_table(table_name: str) → List[Tuple[str, str]]

Function to retrieve all key-value pairs from a particular table in the database.

Parameters:

table_name – Name of the table where the secrets are stored.

Returns:

Returns the secret value.

Return type:

str

vaultapi.database.put_secret(key: str, value: str, table_name: str) → None

Function to add secret to the database.

Parameters:

  • key – Name of the secret to be stored.

  • value – Value of the secret to be stored

  • table_name – Name of the table where the secret is stored.

vaultapi.database.remove_secret(key: str, table_name: str) → None

Function to remove a secret from the database.

Parameters:

  • key – Name of the secret to be removed.

  • table_name – Name of the table where the secret is stored.

vaultapi.database.drop_table(table_name: str) → None

Function to drop a table from the database.

Parameters:

table_name – Name of the table to be dropped.

Exceptions

exception vaultapi.exceptions.APIResponse(status_code: int, detail: Any = None, headers: Optional[Dict[str, str]] = None)

Custom HTTPException from FastAPI to wrap an API response.

>>> APIResponse

Models

class vaultapi.models.RateLimit(BaseModel)

Object to store the rate limit settings.

>>> RateLimit
max_requests: int
seconds: int
class vaultapi.models.Session(BaseModel)

Object to store session information.

>>> Session
fernet: cryptography.fernet.Fernet | None
info: Dict[str, str]
rps: Dict[str, int]
allowed_origins: Set[str]
class Config

Config to allow arbitrary types.

arbitrary_types_allowed = True
class vaultapi.models.EnvConfig(BaseSettings)

Object to load environment variables.

>>> EnvConfig
apikey: str
secret: str
transit_key_length: int
transit_time_bucket: int
database: Union[Path, Path, str]
host: str
port: int
workers: int
log_config: Optional[Union[Path, Dict[str, Any]]]
allowed_origins: Union[Url, List[Url]]
allowed_ip_range: List[str]
rate_limit: Union[RateLimit, List[RateLimit]]
classmethod validate_allowed_origins(value: Union[Url, List[Url]]) → List[Url]

Validate allowed origins to enable CORS policy.

classmethod validate_allowed_ip_range(value: List[str]) → List[str]

Validate allowed IP range to whitelist.

classmethod validate_apikey(value: str) → str | None

Validate API key for complexity.

classmethod validate_api_secret(value: str) → str

Validate API secret to Fernet compatible.

classmethod from_env_file(env_file: Path) → EnvConfig

Create Settings instance from environment file.

Parameters:

env_file – Name of the env file.

Returns:

Loads the EnvConfig model.

Return type:

EnvConfig

class Config

Extra configuration for EnvConfig object.

extra = 'ignore'
hide_input_in_errors = True
arbitrary_types_allowed = True
vaultapi.models.complexity_checker(secret: str) → None

Verifies the strength of a secret.

See also

A secret is considered strong if it at least has:

  • 32 characters

  • 1 digit

  • 1 symbol

  • 1 uppercase letter

  • 1 lowercase letter

Raises:

AssertionError – When at least 1 of the above conditions fail to match.

class vaultapi.models.Database(filepath: Union[Path, str], timeout: int = 10)

Creates a connection and instantiates the cursor.

>>> Database
Parameters:

  • filepath – Name of the database file.

  • timeout – Timeout for the connection to database.

vaultapi.models.database

alias of Database

vaultapi.models.env

alias of EnvConfig

Payload

class vaultapi.payload.DeleteSecret(BaseModel)

Payload for delete-secret API call.

>>> DeleteSecret
key: str
table_name: str
class vaultapi.payload.PutSecret(BaseModel)

Payload for put-secret API call.

>>> PutSecret
key: str
value: str
table_name: str

RateLimit

vaultapi.rate_limit._get_identifier(request: Request) → str

Generate a unique identifier for the request.

class vaultapi.rate_limit.RateLimiter(rps: RateLimit)

Rate limiter for incoming requests.

>>> RateLimiter
init(request: Request) → None

Checks if the number of calls exceeds the rate limit for the given identifier.

Parameters:

request – The incoming request object.

Raises:

429 – Too many requests.

API Routes

async vaultapi.routes.retrieve_secret(key: str, table_name: str) → str | None

Retrieve an existing secret from a table in the database.

Parameters:

  • key – Name of the secret to retrieve.

  • table_name – Name of the table where the secret is stored.

Returns:

Returns the secret value.

Return type:

str

async vaultapi.routes.retrieve_secrets(table_name: str, keys: List[str] = None) → Dict[str, str]

Retrieve multiple secrets from a table or retrieve the table as a whole.

Parameters:

  • table_name – Name of the table where the secret is stored.

  • keys – List of keys for which the values have to be retrieved.

Returns:

Returns the key-value pairs for secret key and it’s value.

Return type:

Dict[str, str]

async vaultapi.routes.get_secret(request: Request, key: str, table_name: str = 'default', apikey: HTTPAuthorizationCredentials = Depends(HTTPBearer))

API function to retrieve a secret.

Args:

request: Reference to the FastAPI request object. key: Name of the secret to be retrieved. table_name: Name of the table where the secret is stored. apikey: API Key to authenticate the request.

Raises:

APIResponse: Raises the HTTPStatus object with a status code and detail as response.

async vaultapi.routes.get_secrets(request: Request, keys: str, table_name: str = 'default', apikey: HTTPAuthorizationCredentials = Depends(HTTPBearer))

API function to retrieve multiple secrets at a time.

Args:

request: Reference to the FastAPI request object. key: Comma separated list of secret names to be retrieved. table_name: Name of the table where the secrets are stored. apikey: API Key to authenticate the request.

Raises:

APIResponse: Raises the HTTPStatus object with a status code and detail as response.

async vaultapi.routes.get_table(request: Request, table_name: str = 'default', apikey: HTTPAuthorizationCredentials = Depends(HTTPBearer))

API function to retrieve ALL the key-value pairs stored in a particular table.

Args:

request: Reference to the FastAPI request object. table_name: Name of the table where the secrets are stored. apikey: API Key to authenticate the request.

Raises:

APIResponse: Raises the HTTPStatus object with a status code and detail as response.

async vaultapi.routes.put_secret(request: Request, data: PutSecret, apikey: HTTPAuthorizationCredentials = Depends(HTTPBearer))

API function to add secrets to database.

Args:

request: Reference to the FastAPI request object. data: Payload with key, value, and table_name as body. apikey: API Key to authenticate the request.

Raises:

APIResponse: Raises the HTTPStatus object with a status code and detail as response.

async vaultapi.routes.put_secrets(request: Request, data: PutSecrets, apikey: HTTPAuthorizationCredentials = Depends(HTTPBearer))

API function to add multiple secrets to a table in the database.

Args:

request: Reference to the FastAPI request object. data: Payload with key, value, and table_name as body. apikey: API Key to authenticate the request.

Raises:

APIResponse: Raises the HTTPStatus object with a status code and detail as response.

async vaultapi.routes.delete_secret(request: Request, data: DeleteSecret, apikey: HTTPAuthorizationCredentials = Depends(HTTPBearer))

API function to delete secrets from database.

Args:

request: Reference to the FastAPI request object. data: Payload with key and table_name as body. apikey: API Key to authenticate the request.

Raises:

APIResponse: Raises the HTTPStatus object with a status code and detail as response.

async vaultapi.routes.create_table(request: Request, table_name: str, apikey: HTTPAuthorizationCredentials = Depends(HTTPBearer))

API function to create a new table in the database.

Args:

request: Reference to the FastAPI request object. table_name: Name of the table to be created. apikey: API Key to authenticate the request.

Raises:

APIResponse: Raises the HTTPStatus object with a status code and detail as response.

async vaultapi.routes.health() → Dict[str, str]

Healthcheck endpoint.

Returns:

Returns the health response.

Return type:

Dict[str, str]

async vaultapi.routes.docs() → RedirectResponse

Redirect to docs page.

Returns:

Redirects the user to /docs page.

Return type:

RedirectResponse

vaultapi.routes.get_all_routes() → List[APIRoute]

Get all the routes to be added for the API server.

Returns:

Returns the routes as a list of APIRoute objects.

Return type:

List[APIRoute]

Squire

vaultapi.squire.envfile_loader(filename: str | os.PathLike) → EnvConfig

Loads environment variables based on filetypes.

Parameters:

filename – Filename from where env vars have to be loaded.

Returns:

Returns a reference to the EnvConfig object.

Return type:

EnvConfig

vaultapi.squire.load_env(**kwargs) → EnvConfig

Merge env vars from env_file with kwargs, giving priority to kwargs.

See also

This function allows env vars to be loaded partially from .env files and partially through kwargs.

Returns:

Returns a reference to the EnvConfig object.

Return type:

EnvConfig

Transit

Module that performs transit encryption/decryption.

This allows the server to securely transmit the retrieved secret to be decrypted at the client side using the API key.

vaultapi.transit.string_to_aes_key(input_string: str, key_length: int) → ByteString

Hashes the string.

Parameters:

  • input_string – String for which an AES hash has to be generated.

  • key_length – AES key size used during encryption.

See also

AES supports three key lengths:

  • 128 bits (16 bytes)

  • 192 bits (24 bytes)

  • 256 bits (32 bytes)

Returns:

Return the first 16 bytes for the AES key

Return type:

str

vaultapi.transit.encrypt(payload: Dict[str, Any], url_safe: bool = True) → Union[ByteString, str]

Encrypt a message using GCM mode with 12 fresh bytes.

Parameters:

  • payload – Payload to be encrypted.

  • url_safe – Boolean flag to perform base64 encoding to perform JSON serialization.

Returns:

Returns the ciphertext as a string or bytes based on the url_safe flag.

Return type:

ByteString | str

vaultapi.transit.decrypt(ciphertext: Union[ByteString, str]) → Dict[str, Any]

Decrypt the ciphertext.

Raises:

Raises InvalidTag if using wrong key or corrupted ciphertext. –

Returns:

Returns the JSON serialized decrypted payload.

Return type:

Dict[str, Any]

Util

vaultapi.util.dotenv_to_table(table_name: str, dotenv_file: str, drop_existing: bool = False, **kwargs) → None

Store all the env vars from a .env file into the database.

Parameters:

  • table_name – Name of the table to store secrets.

  • dotenv_file – Dot env filename.

  • drop_existing – Boolean flag to drop existing table.

vaultapi.util.transit_decrypt(ciphertext: Union[str, ByteString]) → Dict[str, Any]

Decrypts the ciphertext into an appropriate payload.

Parameters:

ciphertext – Encrypted ciphertext.

Returns:

Returns the decrypted payload.

Return type:

Dict[str, Any]

Indices and tables