Log Decryption¶

Native Python decryption for encrypted Silo log entries. Requires pip install -e ".[decrypt]" (from SDK directory) for Standard encryption or .[legacy-decrypt] for seccure/secp256r1 support.

Ciphertext Layout (Standard)¶

Bytes   0–11  : 12-byte GCM nonce (IV)
Bytes  12–102 : 91-byte DER-encoded ephemeral EC public key (P-256 SubjectPublicKeyInfo)
Bytes 103–118 : 16-byte GCM authentication tag
Bytes  119+   : encrypted payload

Key File Format¶

Private keys are stored in pvtkey.txt as name=value pairs:

my_key_name=base64-encoded-private-key-value
another_key=another-base64-encoded-value

Manage keys using scripts/key_manager.py.

Functions¶

Native decryption module for encrypted Silo platform logs.

Provides decryption of log entries encrypted by the Silo platform using either the Standard (EC + HKDF-SHA384 + AES-256-GCM) or Legacy (seccure library, secp256r1) encryption schemes.

Dependencies¶

cryptography>=43.0.0 — required for Standard decryption (install via pip install silo-sdk[decrypt])
seccure>=0.5.0 — required for Legacy decryption (install via pip install silo-sdk[legacy-decrypt])

Key file format¶

Private keys are stored in a simple text file (pvtkey.txt) with one key per line in key-name=key-value format. For Standard encryption the value is the PEM-encoded private key (or a path to a .pem file).

Example::

mykey=-----BEGIN EC PRIVATE KEY-----\nMHQCA...\n-----END EC PRIVATE KEY-----
legacy_key=raw_passphrase_value

Video decryption¶

Encrypted video files can be decrypted with :func:decrypt_video_file, which uses chunked I/O (32 MB chunks by default) to avoid loading the entire file into memory and an atomic write pattern (temp file → move on success).

Decryption failures¶

A tampered/corrupted ciphertext or any other decryption error raises :class:LogDecryptionError — it is never silently reported as a successful decryption. See :func:decrypt_log_entry and :func:decrypt_logs.

LogDecryptionError ¶

Bases: ValueError

Raised when a log entry's ciphertext fails to decrypt or authenticate.

This covers AES-GCM authentication failures (InvalidTag — a tampered or corrupted ciphertext, or the wrong key), malformed ciphertext, and any other error raised while decrypting or JSON-parsing a log entry's plaintext.

Decryption failures are never silently treated as successful: they are never merged into :func:decrypt_log_entry's return value or appended to :func:decrypt_logs's decrypted-entry list. Subclasses ValueError so existing callers that already catch ValueError around decryption calls continue to observe failures.

Source code in silo_sdk/logging/decrypt.py

class LogDecryptionError(ValueError):
    """Raised when a log entry's ciphertext fails to decrypt or authenticate.

    This covers AES-GCM authentication failures (``InvalidTag`` — a
    tampered or corrupted ciphertext, or the wrong key), malformed
    ciphertext, and any other error raised while decrypting or
    JSON-parsing a log entry's plaintext.

    Decryption failures are never silently treated as successful: they
    are never merged into :func:`decrypt_log_entry`'s return value or
    appended to :func:`decrypt_logs`'s decrypted-entry list. Subclasses
    ``ValueError`` so existing callers that already catch ``ValueError``
    around decryption calls continue to observe failures.
    """

load_private_keys ¶

load_private_keys(key_file: str, load_pem: bool = False) -> dict[str, str]

Parse a pvtkey.txt-format key file.

Each non-empty line should be key-name=key-value. When load_pem is True and a value ends with .pem, the value is replaced by the contents of that PEM file.

Values that were base64-encoded by :func:add_key_to_store (detected by successfully base64-decoding to UTF-8 text containing a PEM header) are automatically decoded back to raw PEM text. This makes the generate_key_pair -> add_key_to_store -> load_private_keys -> decrypt workflow work without a manual decoding step. Values that are not base64-encoded PEM (Legacy passphrases, .pem file paths, hand-written raw PEM) are returned unchanged.

Parameters:

Name	Type	Description	Default
`key_file`	`str`	Path to the key file.	required
`load_pem`	`bool`	If `True`, dereference `.pem` file paths.	`False`

Returns:

Type	Description
`dict[str, str]`	Mapping of key name to key data.

Raises:

Type	Description
`FileNotFoundError`	If key_file does not exist.

Source code in silo_sdk/logging/decrypt.py

def load_private_keys(
    key_file: str,
    load_pem: bool = False,
) -> dict[str, str]:
    """Parse a ``pvtkey.txt``-format key file.

    Each non-empty line should be ``key-name=key-value``.  When *load_pem*
    is ``True`` and a value ends with ``.pem``, the value is replaced by the
    contents of that PEM file.

    Values that were base64-encoded by :func:`add_key_to_store` (detected
    by successfully base64-decoding to UTF-8 text containing a PEM header)
    are automatically decoded back to raw PEM text. This makes the
    ``generate_key_pair`` -> ``add_key_to_store`` -> ``load_private_keys``
    -> decrypt workflow work without a manual decoding step. Values that
    are not base64-encoded PEM (Legacy passphrases, ``.pem`` file paths,
    hand-written raw PEM) are returned unchanged.

    Args:
        key_file: Path to the key file.
        load_pem: If ``True``, dereference ``.pem`` file paths.

    Returns:
        Mapping of key name to key data.

    Raises:
        FileNotFoundError: If *key_file* does not exist.
    """
    path = Path(key_file)
    if not path.is_file():
        raise FileNotFoundError(f"Key file not found: {key_file}")

    keys: dict[str, str] = {}
    for line in path.read_text(encoding="utf-8").splitlines():
        line = line.strip()
        if not line or "=" not in line:
            continue
        name, value = line.split("=", 1)
        if load_pem and value.endswith(".pem"):
            pem_path = Path(value)
            if not pem_path.is_file():
                raise FileNotFoundError(f"PEM file not found: {value}")
            value = pem_path.read_text(encoding="utf-8")
        else:
            decoded = _maybe_decode_base64_pem(value)
            if decoded is not None:
                value = decoded
        keys[name] = value
    return keys

decrypt_log_entry ¶

decrypt_log_entry(entry: dict[str, Any], keys: dict[str, str], show_enc_block: bool = False) -> dict[str, Any] | None

Decrypt a single encrypted log entry.

The entry must contain an enc field (base64-encoded ciphertext), a key_name field, and optionally an encryption_type field ("Standard" or "Legacy"; defaults to "Legacy").

Parameters:

Name	Type	Description	Default
`entry`	`dict[str, Any]`	A log entry dict from the API.	required
`keys`	`dict[str, str]`	Mapping of key name → key data (PEM string or passphrase).	required
`show_enc_block`	`bool`	If `True`, keep the original `enc` field.	`False`

Returns:

Type	Description
`dict[str, Any] \| None`	The decrypted entry with `decrypted` fields merged in, or
`dict[str, Any] \| None`	`None` if the required key is missing (`key_name` absent from
`dict[str, Any] \| None`	entry, or not present in keys).

Raises:

Type	Description
`LogDecryptionError`	If decryption or authentication fails for any reason — including a tampered/corrupted ciphertext failing its AES-GCM auth tag (`cryptography.exceptions.InvalidTag`), an unparseable key, or plaintext that isn't valid JSON. This is never swallowed into a success-shaped return value; a failed entry is never mistaken for a genuinely decrypted one.

Source code in silo_sdk/logging/decrypt.py

def decrypt_log_entry(
    entry: dict[str, Any],
    keys: dict[str, str],
    show_enc_block: bool = False,
) -> dict[str, Any] | None:
    """Decrypt a single encrypted log entry.

    The entry must contain an ``enc`` field (base64-encoded ciphertext),
    a ``key_name`` field, and optionally an ``encryption_type`` field
    (``"Standard"`` or ``"Legacy"``; defaults to ``"Legacy"``).

    Args:
        entry: A log entry dict from the API.
        keys: Mapping of key name → key data (PEM string or passphrase).
        show_enc_block: If ``True``, keep the original ``enc`` field.

    Returns:
        The decrypted entry with ``decrypted`` fields merged in, or
        ``None`` if the required key is missing (``key_name`` absent from
        *entry*, or not present in *keys*).

    Raises:
        LogDecryptionError: If decryption or authentication fails for any
            reason — including a tampered/corrupted ciphertext failing
            its AES-GCM auth tag (``cryptography.exceptions.InvalidTag``),
            an unparseable key, or plaintext that isn't valid JSON. This
            is never swallowed into a success-shaped return value; a
            failed entry is never mistaken for a genuinely decrypted one.
    """
    key_name = entry.get("key_name")
    if key_name is None:
        logger.warning("Log entry missing 'key_name' field")
        return None

    key_data = keys.get(key_name)
    if key_data is None:
        logger.debug("Missing key: %s — skipping entry", key_name)
        return None

    encryption_type = entry.get("encryption_type", "Legacy")
    if encryption_type not in ("Standard", "Legacy"):
        raise LogDecryptionError(f"Unsupported encryption_type: {encryption_type!r}")

    raw_enc = entry.get("enc", "")
    ciphertext = base64.b64decode(raw_enc)

    try:
        if encryption_type == "Standard":
            if not _check_cryptography():
                raise ImportError("Standard decryption requires 'cryptography' package")
            from cryptography.hazmat.primitives.serialization import (
                load_pem_private_key,
            )

            private_key = load_pem_private_key(key_data.encode(), None)
            plaintext = standard_decrypt(ciphertext, private_key)
        else:
            plaintext = legacy_decrypt(ciphertext, key_data.encode())

        decrypted = json.loads(plaintext)
    except Exception as exc:
        logger.error(
            "%s decryption failed for key %s (seq_id=%s): %s",
            encryption_type,
            key_name,
            entry.get("seq_id"),
            exc,
        )
        raise LogDecryptionError(
            f"{encryption_type} decryption failed using key {key_name}: {exc}"
        ) from exc

    # Merge decrypted fields into the entry
    result = dict(entry)
    if isinstance(decrypted, dict):
        result.update(decrypted)
    else:
        result["decrypted"] = decrypted

    if not show_enc_block:
        result.pop("enc", None)

    return result

decrypt_logs ¶

decrypt_logs(logs: list[dict[str, Any]], keys: dict[str, str], show_enc_block: bool = False, track_missing: bool = False, track_failed: bool = False) -> list[dict[str, Any]] | tuple[list[dict[str, Any]], list[dict[str, Any]]] | tuple[list[dict[str, Any]], list[dict[str, Any]], list[dict[str, Any]]]

Decrypt a list of encrypted log entries.

Entries whose keys are missing are silently skipped (see track_missing).

Entries that fail decryption or authentication — e.g. a tampered or corrupted ciphertext that fails its AES-GCM auth tag — are never treated as successfully decrypted and are never appended to the decrypted-entry list. By default this function raises :class:LogDecryptionError as soon as such an entry is encountered (stopping the batch); pass track_failed=True to instead collect failed entries into a separate list and keep processing the rest of the batch.

Parameters:

Name	Type	Description	Default
`logs`	`list[dict[str, Any]]`	List of log entry dicts (each with `enc` and `key_name`).	required
`keys`	`dict[str, str]`	Mapping of key name → key data.	required
`show_enc_block`	`bool`	Whether to preserve the `enc` field.	`False`
`track_missing`	`bool`	If `True`, collect entries with missing keys and include `missing_key_entries` in the return value (see Returns). If `False` (default), entries with missing keys are silently skipped and not returned at all.	`False`
`track_failed`	`bool`	If `True`, collect entries that failed decryption or authentication and include `failed_entries` in the return value instead of raising :class:`LogDecryptionError`.	`False`

Returns:

Type	Description
`list[dict[str, Any]] \| tuple[list[dict[str, Any]], list[dict[str, Any]]] \| tuple[list[dict[str, Any]], list[dict[str, Any]], list[dict[str, Any]]]`	`track_missing=False, track_failed=False` (default): list of successfully decrypted log entries.
`list[dict[str, Any]] \| tuple[list[dict[str, Any]], list[dict[str, Any]]] \| tuple[list[dict[str, Any]], list[dict[str, Any]], list[dict[str, Any]]]`	`track_missing=True, track_failed=False`: `(decrypted_logs, missing_key_entries)`.
`list[dict[str, Any]] \| tuple[list[dict[str, Any]], list[dict[str, Any]]] \| tuple[list[dict[str, Any]], list[dict[str, Any]], list[dict[str, Any]]]`	`track_missing=False, track_failed=True`: `(decrypted_logs, failed_entries)`.
`list[dict[str, Any]] \| tuple[list[dict[str, Any]], list[dict[str, Any]]] \| tuple[list[dict[str, Any]], list[dict[str, Any]], list[dict[str, Any]]]`	`track_missing=True, track_failed=True`: `(decrypted_logs, missing_key_entries, failed_entries)`.

Raises:

Type	Description
`LogDecryptionError`	If any entry fails decryption or authentication and track_failed is `False`.

Source code in silo_sdk/logging/decrypt.py

def decrypt_logs(
    logs: list[dict[str, Any]],
    keys: dict[str, str],
    show_enc_block: bool = False,
    track_missing: bool = False,
    track_failed: bool = False,
) -> (
    list[dict[str, Any]]
    | tuple[list[dict[str, Any]], list[dict[str, Any]]]
    | tuple[list[dict[str, Any]], list[dict[str, Any]], list[dict[str, Any]]]
):
    """Decrypt a list of encrypted log entries.

    Entries whose keys are missing are silently skipped (see
    *track_missing*).

    Entries that fail decryption or authentication — e.g. a tampered or
    corrupted ciphertext that fails its AES-GCM auth tag — are never
    treated as successfully decrypted and are never appended to the
    decrypted-entry list. By default this function raises
    :class:`LogDecryptionError` as soon as such an entry is encountered
    (stopping the batch); pass *track_failed=True* to instead collect
    failed entries into a separate list and keep processing the rest of
    the batch.

    Args:
        logs: List of log entry dicts (each with ``enc`` and ``key_name``).
        keys: Mapping of key name → key data.
        show_enc_block: Whether to preserve the ``enc`` field.
        track_missing: If ``True``, collect entries with missing keys and
            include ``missing_key_entries`` in the return value (see
            Returns). If ``False`` (default), entries with missing keys
            are silently skipped and not returned at all.
        track_failed: If ``True``, collect entries that failed decryption
            or authentication and include ``failed_entries`` in the
            return value instead of raising :class:`LogDecryptionError`.

    Returns:
        - ``track_missing=False, track_failed=False`` (default): list of
            successfully decrypted log entries.
        - ``track_missing=True, track_failed=False``:
            ``(decrypted_logs, missing_key_entries)``.
        - ``track_missing=False, track_failed=True``:
            ``(decrypted_logs, failed_entries)``.
        - ``track_missing=True, track_failed=True``:
            ``(decrypted_logs, missing_key_entries, failed_entries)``.

    Raises:
        LogDecryptionError: If any entry fails decryption or
            authentication and *track_failed* is ``False``.
    """
    decrypted: list[dict[str, Any]] = []
    missing_entries: list[dict[str, Any]] = []
    failed_entries: list[dict[str, Any]] = []
    skipped = 0

    for entry in logs:
        try:
            result = decrypt_log_entry(entry, keys, show_enc_block=show_enc_block)
        except LogDecryptionError:
            if not track_failed:
                raise
            logger.error(
                "Entry (key_name=%s, seq_id=%s) failed decryption/authentication "
                "and was excluded from the decrypted list",
                entry.get("key_name"),
                entry.get("seq_id"),
            )
            failed_entries.append(entry)
            continue

        if result is not None:
            decrypted.append(result)
        else:
            skipped += 1
            if track_missing:
                missing_entries.append(entry)

    if skipped:
        logger.info("Skipped %d entries due to missing keys", skipped)
    if failed_entries:
        logger.warning(
            "%d of %d entries failed decryption/authentication and were "
            "excluded from the decrypted list",
            len(failed_entries),
            len(logs),
        )

    if track_missing and track_failed:
        return (decrypted, missing_entries, failed_entries)
    if track_missing:
        return (decrypted, missing_entries)
    if track_failed:
        return (decrypted, failed_entries)
    return decrypted

standard_decrypt ¶

standard_decrypt(ciphertext: bytes, private_key: Any) -> bytes

Decrypt ciphertext produced by Standard (EC-based) encryption.

The ciphertext layout is:

Bytes 0–11 : 12-byte IV (GCM nonce)
Bytes 12–102 : DER-encoded ephemeral EC public key (91 bytes)
Bytes 103–118: 16-byte GCM authentication tag
Bytes 119+ : encrypted payload

Parameters:

Name	Type	Description	Default
`ciphertext`	`bytes`	Raw ciphertext bytes (not base64-encoded).	required
`private_key`	`Any`	A `cryptography` `EllipticCurvePrivateKey` instance.	required

Returns:

Type	Description
`bytes`	Decrypted plaintext bytes.

Raises:

Type	Description
`ImportError`	If `cryptography` is not installed.

Source code in silo_sdk/logging/decrypt.py

def standard_decrypt(ciphertext: bytes, private_key: Any) -> bytes:
    """Decrypt ciphertext produced by Standard (EC-based) encryption.

    The ciphertext layout is:

    * Bytes  0–11  : 12-byte IV (GCM nonce)
    * Bytes 12–102 : DER-encoded ephemeral EC public key (91 bytes)
    * Bytes 103–118: 16-byte GCM authentication tag
    * Bytes 119+   : encrypted payload

    Args:
        ciphertext: Raw ciphertext bytes (not base64-encoded).
        private_key: A ``cryptography`` ``EllipticCurvePrivateKey`` instance.

    Returns:
        Decrypted plaintext bytes.

    Raises:
        ImportError: If ``cryptography`` is not installed.
    """
    if not _check_cryptography():
        raise ImportError(
            "Standard decryption requires the 'cryptography' package. "
            "Install with: pip install 'silo-sdk[decrypt]'"
        )

    from cryptography.hazmat.primitives import hashes, serialization
    from cryptography.hazmat.primitives.asymmetric import ec
    from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
    from cryptography.hazmat.primitives.kdf.hkdf import HKDF

    iv = ciphertext[:12]
    ephemeral_pub_der = ciphertext[12:103]
    tag = ciphertext[103:119]
    encrypted_data = ciphertext[119:]

    ephemeral_key = serialization.load_der_public_key(ephemeral_pub_der)
    shared_key = private_key.exchange(ec.ECDH(), ephemeral_key)
    derived_key = HKDF(
        algorithm=hashes.SHA384(),
        length=32,
        salt=None,
        info=None,
    ).derive(shared_key)

    decryptor = Cipher(
        algorithms.AES(derived_key),
        modes.GCM(iv, tag),
    ).decryptor()
    plaintext: bytes = decryptor.update(encrypted_data) + decryptor.finalize()
    return plaintext

standard_decrypt_chunked ¶

standard_decrypt_chunked(input_file: IO[bytes], output_file: IO[bytes], private_key: Any, chunk_size: int = 32 * 1024 * 1024) -> None

Decrypt a Standard-encrypted file using chunked I/O to reduce memory usage.

Reads the 119-byte header (IV + ephemeral key + auth tag) first, then processes the remaining ciphertext in chunk_size chunks.

.. warning:: AES-GCM authentication is only verified when :py:meth:finalize is called, after every plaintext chunk has already been written to output_file — decryptor.update() returns and writes plaintext immediately, it does not buffer until the auth tag is checked. If the ciphertext has been tampered with or corrupted, :py:meth:finalize raises cryptography.exceptions.InvalidTag, but the unauthenticated plaintext chunks already written to output_file are not automatically removed from it — this function only makes a best-effort attempt to truncate output_file back to its original position when output_file supports seek/truncate (see the except clause below). Callers writing directly to a persistent destination (a real path, not an in-memory buffer) should still write to a temporary file and move it into place only after this function returns successfully — see :func:decrypt_video_file for the reference implementation of that pattern — or must otherwise treat output_file's contents as untrusted and discard them if this function raises.

Parameters:

Name	Type	Description	Default
`input_file`	`IO[bytes]`	Readable binary file object positioned at the start of the encrypted payload.	required
`output_file`	`IO[bytes]`	Writable binary file object for the plaintext output.	required
`private_key`	`Any`	A `cryptography` `EllipticCurvePrivateKey` instance.	required
`chunk_size`	`int`	Bytes to read per iteration (default 32 MB).	`32 * 1024 * 1024`

Raises:

Type	Description
`ImportError`	If `cryptography` is not installed.
`ValueError`	If the file header is too short.
`InvalidTag`	If the ciphertext fails AES-GCM authentication (tampered/corrupted data, or the wrong key). Any bytes already written to output_file before this is raised are unauthenticated and must not be trusted.

Source code in silo_sdk/logging/decrypt.py

def standard_decrypt_chunked(
    input_file: IO[bytes],
    output_file: IO[bytes],
    private_key: Any,
    chunk_size: int = 32 * 1024 * 1024,
) -> None:
    """Decrypt a Standard-encrypted file using chunked I/O to reduce memory usage.

    Reads the 119-byte header (IV + ephemeral key + auth tag) first, then
    processes the remaining ciphertext in *chunk_size* chunks.

    .. warning::
        AES-GCM authentication is only verified when :py:meth:`finalize` is
        called, **after** every plaintext chunk has already been written to
        *output_file* — ``decryptor.update()`` returns and writes plaintext
        immediately, it does not buffer until the auth tag is checked. If
        the ciphertext has been tampered with or corrupted,
        :py:meth:`finalize` raises
        ``cryptography.exceptions.InvalidTag``, but the **unauthenticated**
        plaintext chunks already written to *output_file* are not
        automatically removed from it — this function only makes a
        best-effort attempt to truncate *output_file* back to its
        original position when *output_file* supports ``seek``/``truncate``
        (see the ``except`` clause below). Callers writing directly to a
        persistent destination (a real path, not an in-memory buffer)
        should still write to a temporary file and move it into place only
        after this function returns successfully — see
        :func:`decrypt_video_file` for the reference implementation of
        that pattern — or must otherwise treat *output_file*'s contents as
        untrusted and discard them if this function raises.

    Args:
        input_file: Readable binary file object positioned at the start of
            the encrypted payload.
        output_file: Writable binary file object for the plaintext output.
        private_key: A ``cryptography`` ``EllipticCurvePrivateKey`` instance.
        chunk_size: Bytes to read per iteration (default 32 MB).

    Raises:
        ImportError: If ``cryptography`` is not installed.
        ValueError: If the file header is too short.
        cryptography.exceptions.InvalidTag: If the ciphertext fails
            AES-GCM authentication (tampered/corrupted data, or the wrong
            key). Any bytes already written to *output_file* before this
            is raised are unauthenticated and must not be trusted.
    """
    if not _check_cryptography():
        raise ImportError(
            "Standard decryption requires the 'cryptography' package. "
            "Install with: pip install 'silo-sdk[decrypt]'"
        )

    from cryptography.hazmat.primitives import hashes, serialization
    from cryptography.hazmat.primitives.asymmetric import ec
    from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
    from cryptography.hazmat.primitives.kdf.hkdf import HKDF

    # Header layout: 12 IV + 91 ephemeral DER key + 16 auth tag = 119 bytes
    header = input_file.read(119)
    if len(header) < 119:
        raise ValueError("Invalid encrypted file: header too short")

    iv = header[:12]
    ephemeral_key_der = header[12:103]
    auth_tag = header[103:119]

    ephemeral_key = serialization.load_der_public_key(ephemeral_key_der)
    shared_key = private_key.exchange(ec.ECDH(), ephemeral_key)
    derived_key = HKDF(
        algorithm=hashes.SHA384(),
        length=32,
        salt=None,
        info=None,
    ).derive(shared_key)

    decryptor = Cipher(
        algorithms.AES(derived_key),
        modes.GCM(iv, auth_tag),
    ).decryptor()

    # Best-effort: remember the starting offset so unauthenticated plaintext
    # can be truncated back out of output_file if authentication fails below.
    try:
        start_pos: int | None = output_file.tell()
    except (OSError, AttributeError, io.UnsupportedOperation):  # pragma: no cover
        start_pos = None

    try:
        while True:
            chunk = input_file.read(chunk_size)
            if not chunk:
                break
            output_file.write(decryptor.update(chunk))

        # finalize() validates the auth tag; raises InvalidTag on failure
        final_data = decryptor.finalize()
        if final_data:  # pragma: no branch — AES-GCM finalize returns b"" in practice
            output_file.write(final_data)  # pragma: no cover
    except Exception:
        if start_pos is not None:
            try:
                output_file.seek(start_pos)
                output_file.truncate()
            except (
                OSError,
                AttributeError,
                io.UnsupportedOperation,
            ):  # pragma: no cover
                pass
        raise

decrypt_video_file ¶

decrypt_video_file(video_path: str, output_path: str, key_name: str, key_file: str | None = None, keys: dict[str, str] | None = None) -> bool

Decrypt an encrypted video file using Standard encryption with chunked I/O.

Uses an atomic write pattern: decrypts to a temporary file in the same directory as output_path, then moves it into place on success. Partial files are cleaned up on any failure.

Parameters:

Name	Type	Description	Default
`video_path`	`str`	Path to the encrypted video file.	required
`output_path`	`str`	Destination path for the decrypted output.	required
`key_name`	`str`	Name of the key to use (must exist in the key source).	required
`key_file`	`str \| None`	Path to a `pvtkey.txt`-format key file. Mutually exclusive with keys.	`None`
`keys`	`dict[str, str] \| None`	Pre-loaded key dictionary (alternative to key_file).	`None`

Returns:

Type	Description
`bool`	`True` on success.

Raises:

Type	Description
`ValueError`	If neither key_file nor keys is provided, or if decryption fails (wrong key or corrupt file).
`KeyError`	If key_name is not found in the key source.
`FileNotFoundError`	If video_path does not exist.
`ImportError`	If `cryptography` is not installed.

Source code in silo_sdk/logging/decrypt.py

def decrypt_video_file(
    video_path: str,
    output_path: str,
    key_name: str,
    key_file: str | None = None,
    keys: dict[str, str] | None = None,
) -> bool:
    """Decrypt an encrypted video file using Standard encryption with chunked I/O.

    Uses an atomic write pattern: decrypts to a temporary file in the same
    directory as *output_path*, then moves it into place on success.
    Partial files are cleaned up on any failure.

    Args:
        video_path: Path to the encrypted video file.
        output_path: Destination path for the decrypted output.
        key_name: Name of the key to use (must exist in the key source).
        key_file: Path to a ``pvtkey.txt``-format key file.  Mutually
            exclusive with *keys*.
        keys: Pre-loaded key dictionary (alternative to *key_file*).

    Returns:
        ``True`` on success.

    Raises:
        ValueError: If neither *key_file* nor *keys* is provided, or if
            decryption fails (wrong key or corrupt file).
        KeyError: If *key_name* is not found in the key source.
        FileNotFoundError: If *video_path* does not exist.
        ImportError: If ``cryptography`` is not installed.
    """
    if key_file is None and keys is None:
        raise ValueError("Specify key_file or keys dict to decrypt video")

    if not _check_cryptography():
        raise ImportError(
            "Standard decryption requires the 'cryptography' package. "
            "Install with: pip install 'silo-sdk[decrypt]'"
        )

    from cryptography.exceptions import InvalidTag, UnsupportedAlgorithm
    from cryptography.hazmat.primitives.serialization import load_pem_private_key

    logger.info("Starting video decryption for %s", video_path)

    key_dict = (
        keys if keys is not None else load_private_keys(key_file, load_pem=True)  # type: ignore[arg-type]
    )

    if key_name not in key_dict:
        raise KeyError(f'Key "{key_name}" not found in private key file')

    if not os.path.isfile(video_path):
        raise FileNotFoundError(f"Encrypted video file not found: {video_path}")

    key_data = key_dict[key_name]
    try:
        key_bytes = key_data.encode("utf-8") if isinstance(key_data, str) else key_data
        private_key = load_pem_private_key(key_bytes, None)
    except UnsupportedAlgorithm as exc:
        raise ValueError(f"Unsupported algorithm for key {key_name}: {exc}") from exc
    except ValueError as exc:
        raise ValueError(f"Could not load PEM private key {key_name}: {exc}") from exc

    output_dir = os.path.dirname(os.path.abspath(output_path))
    raw_fd, raw_temp_path = tempfile.mkstemp(dir=output_dir, prefix=".decrypt_")
    temp_fd: int | None = raw_fd
    temp_path: str | None = raw_temp_path
    try:
        with open(video_path, "rb") as in_f:
            with os.fdopen(raw_fd, "wb") as out_f:
                temp_fd = None  # prevent double-close in finally
                standard_decrypt_chunked(in_f, out_f, private_key)

        shutil.move(raw_temp_path, output_path)
        temp_path = None  # successfully moved — skip cleanup
        logger.info("Video decryption complete: %s", output_path)

    except (ValueError, InvalidTag) as exc:
        raise ValueError(
            f'Decryption failed — verify that "{key_name}" is the correct key '
            f"for this video: {exc}"
        ) from exc
    finally:
        if temp_fd is not None:  # pragma: no cover
            try:
                os.close(temp_fd)
            except OSError:
                pass
        if temp_path is not None and os.path.exists(temp_path):
            try:
                os.remove(temp_path)
                logger.debug("Cleaned up partial temp file: %s", temp_path)
            except OSError:  # pragma: no cover
                pass

    return True

legacy_decrypt ¶

legacy_decrypt(ciphertext: bytes, passphrase: bytes) -> bytes

Decrypt ciphertext produced by Legacy (seccure-based) encryption.

Parameters:

Name	Type	Description	Default
`ciphertext`	`bytes`	Raw ciphertext bytes (not base64-encoded).	required
`passphrase`	`bytes`	The passphrase / private key bytes.	required

Returns:

Type	Description
`bytes`	Decrypted plaintext bytes.

Raises:

Type	Description
`ImportError`	If `seccure` is not installed.

Source code in silo_sdk/logging/decrypt.py

def legacy_decrypt(ciphertext: bytes, passphrase: bytes) -> bytes:
    """Decrypt ciphertext produced by Legacy (seccure-based) encryption.

    Args:
        ciphertext: Raw ciphertext bytes (not base64-encoded).
        passphrase: The passphrase / private key bytes.

    Returns:
        Decrypted plaintext bytes.

    Raises:
        ImportError: If ``seccure`` is not installed.
    """
    if not _check_seccure():
        raise ImportError(
            "Legacy decryption requires the 'seccure' package. "
            "Install with: pip install 'silo-sdk[legacy-decrypt]'"
        )

    try:
        import seccure  # noqa: F401
    except ImportError as exc:  # pragma: no cover
        raise ImportError("seccure not available") from exc

    return cast(
        bytes, seccure.decrypt(ciphertext, passphrase, curve="secp256r1/nistp256")
    )