Audit Module API Reference¶

This section documents the internals of the audit module in Bijux CLI.

bijux_cli.services.audit ¶

Provides concrete audit service implementations.

This module defines concrete classes that implement the AuditProtocol. It offers different strategies for handling command auditing and execution, allowing the application to switch between a simulation mode (DryRunAudit) and a real execution mode (RealAudit).

A factory function, get_audit_service, is provided to select the appropriate implementation based on a dry_run flag.

DryRunAudit ¶

DryRunAudit(
    log: ObservabilityProtocol, tel: TelemetryProtocol
)

Bases: _BaseAudit

An audit service that records events and simulates command execution.

Initializes the DryRunAudit service.

Parameters:

log (ObservabilityProtocol) –

The service for structured logging.
tel (TelemetryProtocol) –

The service for event tracking.

Source code in src/bijux_cli/services/audit.py

def __init__(self, log: ObservabilityProtocol, tel: TelemetryProtocol) -> None:
    """Initializes the `DryRunAudit` service.

    Args:
        log (ObservabilityProtocol): The service for structured logging.
        tel (TelemetryProtocol): The service for event tracking.
    """
    super().__init__(log, tel)

cli_audit ¶

cli_audit() -> None

Logs a dry-run CLI audit event.

Source code in src/bijux_cli/services/audit.py

def cli_audit(self) -> None:
    """Logs a dry-run CLI audit event."""
    self._log.log("info", "CLI audit (dry-run)", extra={})
    self._tel.event("audit_cli_dry_run", {})

log ¶

log(cmd: list[str], *, executor: str) -> None

Logs and records a command without executing it.

Parameters:

cmd (list[str]) –

The command and arguments to log.
executor (str) –

The name of the entity executing the command.

Returns:

None ( None ) –

Source code in src/bijux_cli/services/audit.py

def log(self, cmd: list[str], *, executor: str) -> None:
    """Logs and records a command without executing it.

    Args:
        cmd (list[str]): The command and arguments to log.
        executor (str): The name of the entity executing the command.

    Returns:
        None:
    """
    entry = {"cmd": cmd, "executor": executor}
    self._commands.append(entry)
    self._log.log("info", "Dry-run", extra=entry)
    self._tel.event("audit_dry_run", entry)

run ¶

run(
    cmd: list[str], *, executor: str
) -> tuple[int, bytes, bytes]

Simulates the execution of a command.

This method logs the command and returns a successful result without actually running a subprocess.

Parameters:

cmd (list[str]) –

The command to simulate.
executor (str) –

The name of the entity executing the command.

Returns:

tuple[int, bytes, bytes] –

tuple[int, bytes, bytes]: A tuple of dummy values: (0, b"", b"").

Source code in src/bijux_cli/services/audit.py

def run(self, cmd: list[str], *, executor: str) -> tuple[int, bytes, bytes]:
    """Simulates the execution of a command.

    This method logs the command and returns a successful result without
    actually running a subprocess.

    Args:
        cmd (list[str]): The command to simulate.
        executor (str): The name of the entity executing the command.

    Returns:
        tuple[int, bytes, bytes]: A tuple of dummy values: `(0, b"", b"")`.
    """
    self.log(cmd, executor=executor)
    return 0, b"", b""

RealAudit ¶

RealAudit(
    log: ObservabilityProtocol, tel: TelemetryProtocol
)

Bases: _BaseAudit

An audit service that validates, logs, and executes real commands.

Initializes the RealAudit service.

Parameters:

log (ObservabilityProtocol) –

The service for structured logging.
tel (TelemetryProtocol) –

The service for event tracking.

Source code in src/bijux_cli/services/audit.py

def __init__(self, log: ObservabilityProtocol, tel: TelemetryProtocol) -> None:
    """Initializes the `RealAudit` service.

    Args:
        log (ObservabilityProtocol): The service for structured logging.
        tel (TelemetryProtocol): The service for event tracking.
    """
    super().__init__(log, tel)

cli_audit ¶

cli_audit() -> None

Logs a real CLI audit event.

Source code in src/bijux_cli/services/audit.py

def cli_audit(self) -> None:
    """Logs a real CLI audit event."""
    self._log.log(
        "info", "CLI audit (real)", extra={"commands": len(self._commands)}
    )
    self._tel.event("audit_cli_real", {"commands": len(self._commands)})

log ¶

log(cmd: list[str], *, executor: str) -> None

Logs a command with the intent to execute it.

Parameters:

cmd (list[str]) –

The command and arguments to log.
executor (str) –

The name of the entity executing the command.

Returns:

None ( None ) –

Source code in src/bijux_cli/services/audit.py

def log(self, cmd: list[str], *, executor: str) -> None:
    """Logs a command with the intent to execute it.

    Args:
        cmd (list[str]): The command and arguments to log.
        executor (str): The name of the entity executing the command.

    Returns:
        None:
    """
    entry = {"cmd": cmd, "executor": executor}
    self._commands.append(entry)
    self._log.log("debug", f"Executing {executor}", extra=entry)
    self._tel.event("audit_execute", entry)

run ¶

run(
    cmd: list[str], *, executor: str
) -> tuple[int, bytes, bytes]

Validates, logs, and executes a command in a subprocess.

Parameters:

cmd (list[str]) –

The command and arguments to execute.
executor (str) –

The name of the entity executing the command.

Returns:

tuple[int, bytes, bytes] –

tuple[int, bytes, bytes]: A tuple containing the command's return code, standard output, and standard error.

Raises:

BijuxError –

If command validation fails or an unexpected error occurs during execution.

Source code in src/bijux_cli/services/audit.py

def run(self, cmd: list[str], *, executor: str) -> tuple[int, bytes, bytes]:
    """Validates, logs, and executes a command in a subprocess.

    Args:
        cmd (list[str]): The command and arguments to execute.
        executor (str): The name of the entity executing the command.

    Returns:
        tuple[int, bytes, bytes]: A tuple containing the command's return
            code, standard output, and standard error.

    Raises:
        BijuxError: If command validation fails or an unexpected error
            occurs during execution.
    """
    try:
        safe_cmd = validate_command(cmd)
        self.log(safe_cmd, executor=executor)
        proc = subprocess.run(  # noqa: S603 # nosec B603
            safe_cmd,
            capture_output=True,
            check=False,
            shell=False,
        )
        self._tel.event(
            "audit_executed",
            {
                "cmd": safe_cmd,
                "executor": executor,
                "returncode": proc.returncode,
            },
        )
        return proc.returncode, proc.stdout, proc.stderr
    except BijuxError as err:
        self._tel.event(
            "audit_execution_failed",
            {"cmd": cmd, "executor": executor, "error": str(err)},
        )
        raise
    except Exception as err:
        self._tel.event(
            "audit_execution_failed",
            {"cmd": cmd, "executor": executor, "error": str(err)},
        )
        raise BijuxError(f"Failed to execute {executor!r}: {err}") from err

get_audit_service ¶

get_audit_service(
    observability: ObservabilityProtocol,
    telemetry: TelemetryProtocol,
    dry_run: bool = False,
) -> AuditProtocol

A factory function for creating an audit service instance.

Parameters:

observability (ObservabilityProtocol) –

The service for logging.
telemetry (TelemetryProtocol) –

The service for event tracking.
dry_run (bool, default: False ) –

If True, returns a DryRunAudit instance; otherwise, returns a RealAudit instance.

Returns:

AuditProtocol ( AuditProtocol ) –

An instance of the appropriate audit service.

Source code in src/bijux_cli/services/audit.py

def get_audit_service(
    observability: ObservabilityProtocol,
    telemetry: TelemetryProtocol,
    dry_run: bool = False,
) -> AuditProtocol:
    """A factory function for creating an audit service instance.

    Args:
        observability (ObservabilityProtocol): The service for logging.
        telemetry (TelemetryProtocol): The service for event tracking.
        dry_run (bool): If True, returns a `DryRunAudit` instance; otherwise,
            returns a `RealAudit` instance.

    Returns:
        AuditProtocol: An instance of the appropriate audit service.
    """
    return (
        DryRunAudit(observability, telemetry)
        if dry_run
        else RealAudit(observability, telemetry)
    )