Welcome to VaultAPI’s documentation!

Contents:

VaultAPI - Server

vaultapi.server.start() → None

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

async vaultapi.api.delete_ui_session(event: str) → None

Clear any existing UI sessions, so the tokens can’t be re-used when the server is restarted.

vaultapi.api.lifespan(_: FastAPI)

Lifespan context manager.

vaultapi.api.startup() → None

Enables CORS policy and API and UI routes.

Authenticator

vaultapi.auth.UI_BASIC(session, auth, host)
vaultapi.auth.API_BASIC(auth)
vaultapi.auth.API_ADVANCED(auth)
class vaultapi.auth.AuthType(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Model for the authentication type.

>>> AuthType
ui_basic = 'UI_BASIC'
ui_login = 'UI_LOGIN'
ui_advanced = 'UI_ADVANCED'
api_basic = 'API_BASIC'
api_advanced = 'API_ADVANCED'
vaultapi.auth.unauthorized(host: str) → NoReturn

Raise a 403 APIResponse if the host is unauthorized.

async vaultapi.auth.blocked(host: str) → Union[None, NoReturn]

Raise a 403 APIResponse if the host is within an active cool-off window.

Parameters:

host – Hostname or IP address of the client.

Raises:

  • APIResponse –

  • - 403 – If the host has a non-expired blocked_until entry.

async vaultapi.auth.validate_totp(totp_code: str) → Union[bool, NoReturn]

Validate the login credentials from the request body.

Parameters:

totp_code – TOTP code received from the client.

Returns:

True if totp code is valid, False otherwise.

Return type:

bool

async vaultapi.auth.validate(request: Request, authorization: HTTPAuthorizationCredentials, auth_type: AuthType) → Union[None, NoReturn]

Validates the auth request using HTTPBearer.

Parameters:

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

  • authorization – Basic APIKey required for API routes [OR] session token required for UI routes.

  • auth_type – The type of authentication to use.

Raises:

  • APIResponse –

  • - 401 – If authorization is invalid.

  • - 403 – If host address is forbidden (blocked after repeated failures).

Database

vaultapi.database.create_auth_tables() → None

Create auth-specific tables in auth.db if they do not already exist.

vaultapi.database.upsert_ui_session(token: str, hostname: str, expires: int, fernet: Fernet) → None

Persist an active UI session, replacing any previous one.

The stored blob is fernet.encrypt(json({"token": ..., "host": ..., "exp": ...})). Fernet provides authenticated encryption — tampering with the blob is detected on decrypt and raises an exception, which get_ui_session treats as “no valid session”.

Parameters:

  • token – The opaque session token returned to the browser.

  • hostname – request.client.host captured at login time.

  • expires – Unix timestamp after which the session is invalid.

  • fernet – Fernet instance from models.session.fernet.

vaultapi.database.get_ui_session(fernet: Fernet) → dict | None

Return the decrypted session record, or None if absent or tampered.

Parameters:

fernet – Fernet instance from models.session.fernet.

Returns:

{"token": str, "host": str, "exp": int} on success, None otherwise.

Return type:

dict

vaultapi.database.delete_ui_session() → None

Remove all rows from the ui_session table, invalidating any active session.

vaultapi.database.get_blocked_until(host: str) → vaultapi.models.AuthCounter | None

Return the blocked_until Unix timestamp for a host, or None if not blocked.

If the entry exists but its cooloff has expired, it is removed and None is returned.

Parameters:

host – Client IP address to check.

Returns:

Returns a reference to the AuthCounter model if a host has been blocked.

Return type:

models.AuthCounter | None

vaultapi.database.is_host_blocked(host: str) → bool

Return True if the host is currently within a cooloff window.

Parameters:

host – Client IP address to check.

vaultapi.database.remove_blocked_host(host: str) → None

Delete a host’s entry from the blocked_hosts table entirely.

Parameters:

host – Client IP address to remove.

vaultapi.database.increment_failed_auth(host: str) → None

Increment the failed authentication counter for a host and set cooloff if a threshold is crossed.

Inserts the host with count 1 if it does not exist yet. Sets blocked_until to int(time.time()) + duration when the cumulative count reaches 3, 5, or 10.

Parameters:

host – Client IP address that failed authentication.

vaultapi.database.reset_failed_auth(host: str) → None

Reset the failed authentication counter for a host after a successful login.

Parameters:

host – Client IP address to reset.

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.list_tables() → List[str]

Function to list all available tables in the database.

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, bytes]]

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.

vaultapi.database.rename_table(old_name: str, new_name: str) → None

Function to rename a table in the database.

Parameters:

  • old_name – Current name of the table.

  • new_name – New name for the table.

Exceptions

exception vaultapi.exceptions.APIResponse(status_code: int, detail: Any = None, headers: collections.abc.Mapping[str, str] | None = None)

Custom HTTPException from FastAPI to wrap an API response.

>>> APIResponse
exception vaultapi.exceptions.StartupError

Custom StartupError to indicate an error during application startup.

>>> StartupError

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]
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]
auth_database: Union[Path, Path, str]
host: str
port: int
workers: int
enable_ui: bool
totp_token: str | None
ui_lifetime: int
log_config: Optional[Union[Path, Dict[str, Any]]]
allowed_origins: Union[HttpUrl, List[HttpUrl]]
rate_limit: Union[RateLimit, List[RateLimit]]
classmethod validate_transit_key_length(value: int) → Union[int, NoReturn]

Validate transit key length.

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, max_len: int = 32) → 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.

vaultapi.models.validate_totp_secret(token) → Union[None, NoReturn]

Validate the provided TOTP secret token.

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.

class vaultapi.models.AuthCounter(*, count: int, blocked_until: int)

Base model for auth counters.

>>> AuthCounter
count: int
blocked_until: int
_abc_impl = <_abc._abc_data object>
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

vaultapi.models.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.models.load_env() → EnvConfig

Loads te env vars based on the env_file provided.

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

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-secrets API call.

>>> PutSecret
secrets: Union[Dict[str, str], 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

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

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

Returns:

Returns the UI routes as a list of APIRoute objects.

Return type:

List[APIRoute]

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

Get the API routes to be added for the server.

Returns:

Returns the routes as a list of APIRoute objects.

Return type:

List[APIRoute]

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) → 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