tiny_ca.db.sync_db_manager module

Synchronous SQLAlchemy-backed implementation of the certificate registry.

Module-level contents

DatabaseManager — owns the SQLAlchemy engine and session factory.

No domain logic; connection lifecycle only.

SyncDBHandler — implements BaseDB; all certificate registry

operations with explicit, atomic transaction management.

SOLID notes

SRPDatabaseManager handles engine creation and session provisioning.

SyncDBHandler handles domain operations on CertificateRecord rows. Neither class knows about the other’s internal details.

OCPAdditional query methods are added to SyncDBHandler without touching

DatabaseManager or BaseDB.

LSPSyncDBHandler satisfies the full BaseDB contract and is

transparently substitutable in any consumer that depends on BaseDB.

ISPBaseDB is declared in db/base_db.py; this module imports and

extends it without adding unrelated methods visible to callers.

DIPCertLifecycleManager depends on BaseDB, never on

SyncDBHandler directly.

class tiny_ca.db.sync_db_manager.DatabaseManager(db_url='sqlite:///ca_repository.db', create_all=True)

Bases: object

Manages the SQLAlchemy engine and sessionmaker factory.

Responsibility: connection lifecycle only. This class creates the engine, optionally initialises the schema, and hands out new Session objects on request. It performs no queries and contains no domain logic.

Parameters:

• db_url (str) – SQLAlchemy database URL. Supports any SQLAlchemy-compatible backend, e.g. "sqlite:///ca.db" or "postgresql+psycopg2://user:pass@host/dbname". Default: "sqlite:///ca_repository.db".

• create_all (bool) – When True (default), calls Base.metadata.create_all at construction so the schema is present before the first query. Set to False when using Alembic or another migration tool that manages the schema independently.

__init__(db_url='sqlite:///ca_repository.db', create_all=True)

Parameters:

• db_url (str)

• create_all (bool)

Return type:

None

session()

Provision a new SQLAlchemy Session bound to the managed engine.

The caller is fully responsible for the session lifecycle: committing successful transactions, rolling back on errors, and closing the session in a finally block. Example:

session = db_manager.session()
try:
    session.add(record)
    session.commit()
except Exception:
    session.rollback()
    raise
finally:
    session.close()

Returns:

A new, uncommitted SQLAlchemy ORM session.

Return type:

Session

class tiny_ca.db.sync_db_manager.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