tiny_ca.db package

class tiny_ca.db.BaseDB

Bases: ABC

Abstract contract for the certificate registry database adapter.

Defines all operations that the application layer requires from the persistence tier. Concrete implementations (e.g. SyncDBHandler) provide the actual SQL queries and transaction management.

All methods that return a CertificateRecord must return None (not raise an exception) when the requested record simply does not exist. Exceptions are reserved for genuine infrastructure failures (connection errors, constraint violations, etc.).

abstractmethod delete_by_uuid(uuid)

Permanently delete the certificate record identified by uuid.

This is a hard delete — the row is removed from the database. The caller is responsible for also removing the corresponding filesystem artefacts via BaseStorage.delete_certificate_folder.

Parameters:

uuid (str) – The storage folder UUID that uniquely identifies the certificate.

Returns:

True if a row was found and deleted; False if no matching record existed or if a database error occurred.

Return type:

bool

abstractmethod get_by_name(common_name)

Retrieve the currently active certificate record for a given Common Name.

Implementations must filter to only VALID records so that the caller always receives the live certificate or None — never a revoked or expired one.

Parameters:

common_name (str) – The CN (Common Name) field from the certificate Subject to look up.

Returns:

The active VALID record for common_name, or None if no such record exists.

Return type:

CertificateRecord | None

abstractmethod get_by_serial(serial)

Retrieve a certificate record by its X.509 serial number.

Parameters:

serial (int) – Integer serial number to look up. Implementations are responsible for any type conversion required by the underlying storage format (e.g. converting to str for string-typed database columns).

Returns:

The matching record regardless of its current status (VALID, REVOKED, EXPIRED), or None if no record exists for serial.

Return type:

CertificateRecord | None

abstractmethod get_expiring(within_days=30)

Return VALID certificates that expire within within_days calendar days.

Only records with status == VALID are considered — already-revoked or expired records are excluded.

Parameters:

within_days (int) – Look-ahead window in calendar days. Default: 30.

Returns:

Records ordered by not_valid_after ascending (soonest first). Returns an empty list on error.

Return type:

list[CertificateRecord]

abstractmethod get_revoked_certificates()

Yield certificate records that should appear in the current CRL.

Implementations define their own freshness window (e.g. only records revoked within the past 365 days and not yet expired), but must yield objects that expose at minimum:

• serial_number — castable to int

• revocation_date — a datetime object

• revocation_reason — an integer RFC 5280 reason code

Yields:

CertificateRecord – Records (or row-like objects) for each revoked certificate that falls within the implementation’s CRL inclusion window.

Return type:

Generator[CertificateRecord, None, None]

abstractmethod list_all(status=None, key_type=None, limit=100, offset=0)

Return a paginated list of certificate records with optional filters.

Parameters:

• status (str | None) – Filter by lifecycle state ("valid", "revoked", "expired"). None returns all statuses.

• key_type (str | None) – Filter by certificate category ("ca", "service", "device", etc.). None returns all types.

• limit (int) – Maximum number of records to return. Default: 100.

• offset (int) – Number of records to skip (for pagination). Default: 0.

Returns:

Matching records ordered by id descending (newest first). Returns an empty list on error.

Return type:

list[CertificateRecord]

abstractmethod register_cert_in_db(cert, uuid, key_type=CertType.DEVICE)

Persist a newly issued certificate to the registry.

Creates a new record with status=VALID populated from the certificate metadata. The implementation must extract the CN from cert.subject and store the full PEM encoding for later retrieval.

Parameters:

• cert (x509.Certificate) – The issued X.509 certificate object to register.

• uuid (str) – UUID that identifies the filesystem folder containing the corresponding .pem, .key, and .csr artefact files.

• key_type (CertType) – Certificate category (CA, USER, SERVICE, DEVICE, INTERNAL). Default: CertType.DEVICE.

Returns:

True if the record was persisted successfully; False if the operation failed (the implementation must log the reason and roll back any partial changes).

Return type:

bool

abstractmethod revoke_certificate(serial_number, reason=<ReasonFlags.unspecified: 'unspecified'>)

Mark a certificate as revoked and record the reason and timestamp.

Implementations must: 1. Look up the record by serial_number filtered to status=VALID. 2. If not found, return (False, <NOT_FOUND status>) without error. 3. Update status, revocation_reason, and revocation_date. 4. Commit atomically; roll back on any exception.

Parameters:

• serial_number (int) – Serial number of the certificate to revoke.

• reason (x509.ReasonFlags) – RFC 5280 §5.3.1 revocation reason code. Default: x509.ReasonFlags.unspecified (code 0).

Returns:

(success, status_value) where status_value is implementation- defined (typically a RevokeStatus enum member). True indicates the revocation was committed; False means it was not (reason encoded in status_value).

Return type:

tuple[bool, object]

abstractmethod update_status_expired()

Bulk-update all VALID certificates whose not_valid_after has passed.

Sets status = "expired" for every record where: - status == "valid" - not_valid_after < now (UTC)

This method is intended to be called periodically by a background task (e.g. a cron job or an APScheduler job) so that status queries reflect reality without per-request date comparisons.

Returns:

Number of rows updated. Returns 0 on error (after logging).

Return type:

int

class tiny_ca.db.CertificateRecord(**kwargs)

Bases: Base

ORM model for a single certificate entry in the registry.

Maps to the certificates table. Each row represents one certificate that has been issued by the CA, regardless of its current lifecycle state.

Columns

idint

Auto-incremented surrogate primary key. Not exposed to application code; use serial_number as the business key.

serial_numberstr

X.509 serial number stored as a decimal string. Unique and indexed. String storage avoids integer overflow for 160-bit serials (RFC 5280).

common_namestr

Common Name (CN) extracted from the certificate Subject at issuance time. Not unique; the same CN may appear across different certificate generations (e.g. after rotation).

statusstr

Current lifecycle state. One of the CertificateStatus values: "valid", "revoked", "expired", or "unknown". Defaults to CertificateStatus.VALID on insertion.

not_valid_beforedatetime

Start of the certificate’s validity period (UTC, naive datetime as stored by SQLAlchemy’s DateTime Column type).

not_valid_afterdatetime

End of the certificate’s validity period (UTC, naive datetime). Indexed to allow efficient queries for expired certificates.

key_typestr

Certificate category stored as the CertType enum’s string value (e.g. "ca", "device", "service"). Defaults to CertType.DEVICE.value.

certificate_pemstr

Full PEM-encoded public certificate. Allows reconstruction of the x509.Certificate object without accessing the filesystem.

revocation_datedatetime | None

UTC timestamp at which the certificate was revoked. None for non-revoked certificates.

revocation_reasonint | None

RFC 5280 §5.3.1 revocation reason code stored as an integer. None for non-revoked certificates. Maps to the integer value of the corresponding x509.ReasonFlags member.

uuidstr | None

UUID string that identifies the filesystem folder (managed by BaseStorage) holding the .pem, .key, and .csr files for this certificate. None if no filesystem artefacts exist.

A simple constructor that allows initialization from kwargs.

Sets attributes on the constructed instance using the names and values in kwargs.

Only keys that are present as attributes of the instance’s class are allowed. These could be, for example, any mapped columns or relationships.

__init__(**kwargs)

A simple constructor that allows initialization from kwargs.

Sets attributes on the constructed instance using the names and values in kwargs.

Only keys that are present as attributes of the instance’s class are allowed. These could be, for example, any mapped columns or relationships.

certificate_pem: Mapped[str]

common_name: Mapped[str]

id: Mapped[int]

key_type: Mapped[str]

not_valid_after: Mapped[datetime]

not_valid_before: Mapped[datetime]

revocation_date: Mapped[datetime]

revocation_reason: Mapped[str]

serial_number: Mapped[str]

status: Mapped[str]

uuid: Mapped[str]

class tiny_ca.db.CertificateStatus(value)

Bases: StrEnum

Lifecycle state of a certificate record in the registry.

Stored as a lowercase string in the status column of CertificateRecord so the value is human-readable in raw SQL output.

Members

VALID :

The certificate was issued successfully and has not been revoked or expired. Active certificates used for authentication or encryption are expected to be in this state.

REVOKED :

The certificate was explicitly revoked before its natural expiry. The revocation_date and revocation_reason columns on the corresponding CertificateRecord row must be non-null.

EXPIRED :

The certificate’s not_valid_after date has passed. This status may be set by a background job; alternatively, callers can detect expiry by comparing not_valid_after to the current time.

UNKNOWN :

The status could not be determined, typically because no record was found for the requested serial number. Used as a safe sentinel value by CertLifecycleManager.get_certificate_status.

EXPIRED = 'expired'

REVOKED = 'revoked'

UNKNOWN = 'unknown'

VALID = 'valid'

class tiny_ca.db.RevokeStatus(value)

Bases: Enum

Enumerated outcomes for a certificate revocation attempt.

Used as the second element of the (bool, RevokeStatus) tuple returned by BaseDB.revoke_certificate implementations. The boolean indicates overall success; this enum provides the machine-readable reason when the operation did not succeed, and a confirmation token when it did.

Members

NOT_FOUND :

No active (VALID) certificate with the requested serial number exists in the registry. The certificate may already be revoked, expired, or was never registered.

UNKNOWN_ERROR :

An unexpected internal error (e.g. database constraint violation, connection failure) prevented the revocation. The implementation must log the underlying exception before returning this status.

OK :

The revocation was committed successfully. The certificate record has been updated with status=REVOKED, a revocation_date, and the provided revocation_reason.

Examples

>>> success, status = db.revoke_certificate(serial=12345, reason=ReasonFlags.key_compromise)
>>> if not success:
...     if status == RevokeStatus.NOT_FOUND:
...         logger.warning("Certificate not found")
...     else:
...         logger.error("Internal revocation error")

NOT_FOUND = 'The certificate was not revoked because there is no valid certificate with the specified serial number.'

OK = 'success'

UNKNOWN_ERROR = 'The certificate was not revoked due to an internal error. Please review the service logs.'

class tiny_ca.db.SyncDBHandler(db_url, logger=None)

Bases: BaseDB

Synchronous, SQLAlchemy-backed certificate registry.

Implements the full BaseDB contract with explicit, atomic transaction management. Every public method follows the same pattern:

1. Open a new session via self._db.session().

2. Execute the query / mutation inside a try block.

3. Commit on success or roll back on any exception.

4. Always close the session in the finally block.

This guarantees that no session is leaked regardless of outcome, and that partial writes are never visible to other readers.

Parameters:

• db_url (str) – SQLAlchemy database URL forwarded to DatabaseManager.

• logger (Logger | None) – Logger for operational and diagnostic messages. Falls back to DEFAULT_LOGGER when None.

__init__(db_url, logger=None)

Parameters:

• db_url (str)

• logger (Logger | None)

Return type:

None

delete_by_uuid(uuid)

Permanently delete the certificate record identified by uuid.

Parameters:

uuid (str) – Storage folder UUID of the certificate to delete.

Returns:

True if a row was deleted; False otherwise.

Return type:

bool

get_by_name(common_name)

Fetch the active VALID certificate record for the given Common Name.

Only records with status == CertificateStatus.VALID are returned. Revoked and expired records are ignored so that the caller always receives the currently-active certificate for a given CN, or None if no active certificate exists.

Parameters:

common_name (str) – The CN (Common Name) value from the certificate Subject field.

Returns:

The matching VALID record, or None if absent or on DB error.

Return type:

CertificateRecord | None

get_by_serial(serial)

Fetch a single certificate record by its X.509 serial number.

The serial is stored as a string in the database (to avoid integer overflow across all backends); the conversion is handled internally.

Parameters:

serial (int) – Integer serial number to look up.

Returns:

The matching ORM record, or None if no record exists for serial or if a database error occurs.

Return type:

CertificateRecord | None

get_expiring(within_days=30)

Return VALID certificates expiring within within_days calendar days.

Parameters:

within_days (int) – Look-ahead window in days. Default: 30.

Returns:

Records ordered by not_valid_after ascending. Empty list on error.

Return type:

list[CertificateRecord]

get_revoked_certificates()

Yield revoked certificate rows relevant for the current CRL window.

A record is included when all of the following conditions hold:

1. revocation_date is not NULL — the certificate was actually revoked.

2. not_valid_after > now — the certificate has not yet expired; expired certificates need not appear in a CRL because relying parties will reject them regardless.

3. revocation_date > now - 365 days — the revocation is recent enough to be relevant. This prevents unbounded CRL growth from very old entries that no relying party could still encounter.

Only three columns are selected (serial_number, revocation_date, revocation_reason) to minimise data transfer; callers must not access other CertificateRecord attributes on the yielded rows.

Yields:

CertificateRecord – SQLAlchemy Row objects with serial_number, revocation_date, and revocation_reason attributes.

Return type:

Generator[CertificateRecord, None, None]

Notes

All rows are fetched in a single query before yielding begins. The session is closed in the finally block; do not use the yielded rows after the generator has been exhausted or abandoned.

Return type:

Generator[CertificateRecord, None, None]

list_all(status=None, key_type=None, limit=100, offset=0)

Return a paginated list of certificate records with optional filters.

Parameters:

• status (str | None) – Filter by lifecycle state. None returns all statuses.

• key_type (str | None) – Filter by certificate category. None returns all types.

• limit (int) – Maximum number of records to return. Default: 100.

• offset (int) – Number of records to skip (pagination). Default: 0.

Returns:

Records ordered by id descending. Empty list on error.

Return type:

list[CertificateRecord]

register_cert_in_db(cert, uuid, key_type=CertType.DEVICE)

Persist a newly issued certificate to the registry.

Creates a new CertificateRecord row with status=VALID from the metadata and PEM encoding of cert. The full PEM is stored so the certificate can be reconstructed independently of the filesystem.

Parameters:

• cert (x509.Certificate) – The issued X.509 certificate. Its Subject must contain at least one commonName (CN) attribute.

• uuid (str) – UUID string that identifies the storage folder holding the corresponding .pem, .key, and .csr files.

• key_type (CertType) – Certificate category. Stored as its str value (e.g. "device"). Default: CertType.DEVICE.

Returns:

True if the record was committed successfully; False if the operation was rolled back due to an error (e.g. duplicate serial, constraint violation).

Return type:

bool

Raises:

IndexError – Re-raised if the certificate contains no CN attribute, indicating a malformed certificate that should not be stored.

revoke_certificate(serial_number, reason=<ReasonFlags.unspecified: 'unspecified'>)

Mark a certificate as revoked and record the reason and timestamp.

Looks up the certificate by serial_number filtered to status == VALID — already-revoked or unknown serials are treated as not found. On success the record is updated in-place: - status → CertificateStatus.REVOKED - revocation_reason → integer value of reason - revocation_date → current UTC timestamp

The change is committed atomically; a rollback is performed on any unexpected error.

Parameters:

• serial_number (int) – Serial number of the certificate to revoke.

• reason (x509.ReasonFlags) – RFC 5280 §5.3.1 revocation reason code. Default: x509.ReasonFlags.unspecified (code 0).

Returns:

(True,  RevokeStatus.OK) — revocation committed. (False, RevokeStatus.NOT_FOUND) — no VALID cert with that serial. (False, RevokeStatus.UNKNOWN_ERROR) — unexpected internal error.

Return type:

tuple[bool, RevokeStatus]

update_status_expired()

Bulk-set status=expired for all VALID certs whose validity has passed.

Returns:

Number of rows updated. 0 on error.

Return type:

int

tiny_ca.db.SyncDatabaseManager

alias of DatabaseManager

Submodules

• tiny_ca.db.async_db_manager module
  • AsyncDBHandler
    • AsyncDBHandler.__init__()
    • AsyncDBHandler.delete_by_uuid()
    • AsyncDBHandler.get_by_name()
    • AsyncDBHandler.get_by_serial()
    • AsyncDBHandler.get_expiring()
    • AsyncDBHandler.get_revoked_certificates()
    • AsyncDBHandler.list_all()
    • AsyncDBHandler.register_cert_in_db()
    • AsyncDBHandler.revoke_certificate()
    • AsyncDBHandler.update_status_expired()
  • DatabaseManager
    • DatabaseManager.__init__()
    • DatabaseManager.get_session()
    • DatabaseManager.init_db()
• tiny_ca.db.base_db module
  • SOLID notes
  • BaseDB
    • BaseDB.delete_by_uuid()
    • BaseDB.get_by_name()
    • BaseDB.get_by_serial()
    • BaseDB.get_expiring()
    • BaseDB.get_revoked_certificates()
    • BaseDB.list_all()
    • BaseDB.register_cert_in_db()
    • BaseDB.revoke_certificate()
    • BaseDB.update_status_expired()
• tiny_ca.db.const module
  • CertificateStatus
    • CertificateStatus.EXPIRED
    • CertificateStatus.REVOKED
    • CertificateStatus.UNKNOWN
    • CertificateStatus.VALID
  • RevokeStatus
    • RevokeStatus.NOT_FOUND
    • RevokeStatus.OK
    • RevokeStatus.UNKNOWN_ERROR
• tiny_ca.db.models module
  • Module-level contents
  • Design notes
  • CertificateRecord
    • CertificateRecord.__init__()
    • CertificateRecord.certificate_pem
    • CertificateRecord.common_name
    • CertificateRecord.id
    • CertificateRecord.key_type
    • CertificateRecord.not_valid_after
    • CertificateRecord.not_valid_before
    • CertificateRecord.revocation_date
    • CertificateRecord.revocation_reason
    • CertificateRecord.serial_number
    • CertificateRecord.status
    • CertificateRecord.uuid
• tiny_ca.db.sync_db_manager module
  • Module-level contents
  • SOLID notes
  • DatabaseManager
    • DatabaseManager.__init__()
    • DatabaseManager.session()
  • SyncDBHandler
    • SyncDBHandler.__init__()
    • SyncDBHandler.delete_by_uuid()
    • SyncDBHandler.get_by_name()
    • SyncDBHandler.get_by_serial()
    • SyncDBHandler.get_expiring()
    • SyncDBHandler.get_revoked_certificates()
    • SyncDBHandler.list_all()
    • SyncDBHandler.register_cert_in_db()
    • SyncDBHandler.revoke_certificate()
    • SyncDBHandler.update_status_expired()