GeezerVison

- New `--upload-logs` flag (default off). When set, each CLI invocation registers a run, reports a per-run status (`in_progress` / `success` / `failure` / `cancelled`), and uploads a transcript of its own log output to the Socket backend for that run, visible in the Socket admin views. The transcript is captured regardless of the local `--enable-debug` state; the existing terminal verbosity is unchanged.

- New `--no-upload-logs` flag (mutually exclusive with `--upload-logs`) explicitly opts the run out of uploading logs, even when an org-level override would otherwise enable it. Use this when you need a guaranteed no-upload guarantee (e.g. legal/consent reasons).

- The Socket backend can also force-enable streaming for specific orgs in the absence of an explicit opt-out. The feature is best-effort — registration or upload failures silently degrade and never block the scan.

[project]

name = "socketsecurity"

version = "2.4.8"

version = "2.4.9"

requires-python = ">= 3.11"

license = {"file" = "LICENSE"}

dependencies = [

__author__ = 'socket.dev'

__version__ = '2.4.8'

__version__ = '2.4.9'

USER_AGENT = f'SocketPythonCLI/{__version__}'

ignore_commit_files: bool = False

disable_blocking: bool = False

disable_ignore: bool = False

# Tri-state log-upload preference: True = --upload-logs, False = --no-upload-logs,

# None = neither (server-side override decides).

upload_logs: Optional[bool] = None

strict_blocking: bool = False

integration_type: IntegrationType = "api"

integration_org_slug: Optional[str] = None

'ignore_commit_files': args.ignore_commit_files,

'disable_blocking': args.disable_blocking,

'disable_ignore': args.disable_ignore,

'upload_logs': args.upload_logs,

'strict_blocking': args.strict_blocking,

'integration_type': args.integration,

'pending_head': args.pending_head,

action="store_true",

help=argparse.SUPPRESS

)

log_upload_group = advanced_group.add_mutually_exclusive_group()

log_upload_group.add_argument(

"--upload-logs",

dest="upload_logs",

action="store_const",

const=True,

help="Upload the CLI's log output to the Socket backend for this run. "

"When set, the CLI registers the run with share_logs=true and streams "

"its log records in 5s batches. Default off. Mutually exclusive with "

"--no-upload-logs."

)

log_upload_group.add_argument(

"--no-upload-logs",

dest="upload_logs",

action="store_const",

const=False,

help="Explicitly opt out of uploading CLI logs to the Socket backend, even "

"when an org-level override would otherwise enable it. Mutually "

"exclusive with --upload-logs."

)

advanced_group.add_argument(

"--strict-blocking",

dest="strict_blocking",

import json

import logging

from typing import Optional

from .cli_client import CliClient

log = logging.getLogger("socketcli")

def register_cli_run(

client: CliClient,

client_version: str,

upload_logs: Optional[bool],

) -> Optional[str]:

"""Register a CLI run with the backend.

try:

resp = client.request(

path="python-cli-runs",

method="POST",

payload=json.dumps({

"client_version": client_version,

"share_logs": upload_logs is True,

"decline_logs": upload_logs is False,

}),

)

body = resp.json()

if not body.get("log_streaming_enabled"):

log.debug("cli-run register: log streaming not enabled by server")

return None

run_id = body.get("run_id")

if not isinstance(run_id, str) or not run_id:

log.debug(f"cli-run register: enabled but missing run_id in response: {body!r}")

return None

return run_id

except Exception as e:

log.debug(f"cli-run register failed (streaming disabled): {e}")

return None

def finalize_cli_run(

client: CliClient,

run_id: str,

status: str = "success",

report_run_id: Optional[str] = None,

) -> None:

try:

client.request(

path=f"python-cli-runs/{run_id}/finalize",

method="POST",

payload=json.dumps({"status": status, "report_run_id": report_run_id}),

)

except Exception as e:

log.debug(f"cli-run finalize failed (swallowed): {e}")

import json

import logging

import threading

from datetime import datetime, timezone

from typing import Optional

from .cli_client import CliClient

log = logging.getLogger(__name__)

_FLUSH_GUARD = threading.local()

def _now_str() -> str:

return datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]

class BatchedLogUploader:

def __init__(

self,

client: CliClient,

run_id: str,

flush_interval: float = 5.0,

):

self._client = client

self._run_id = run_id

self._flush_interval = flush_interval

self._buf: list = []

self._lock = threading.Lock()

self._stop = threading.Event()

self._thread: Optional[threading.Thread] = None

def add(self, entry: dict) -> None:

with self._lock:

self._buf.append(entry)

def start(self) -> None:

if self._thread is not None:

return

self._thread = threading.Thread(

target=self._run,

name=f"socket-log-uploader-{self._run_id[:8]}",

daemon=True,

)

self._thread.start()

def stop(self, timeout: float = 2.0) -> None:

if self._thread is not None:

self._stop.set()

self._thread.join(timeout=timeout)

self._thread = None

self._flush()

def _run(self) -> None:

while not self._stop.is_set():

self._flush()

self._stop.wait(self._flush_interval)

def _flush(self) -> None:

with self._lock:

if not self._buf:

return

batch = self._buf

self._buf = []

_FLUSH_GUARD.active = True

try:

self._client.request(

path=f"python-cli-runs/{self._run_id}/logs",

method="POST",

payload=json.dumps({"logs": batch}),

)

except Exception as e:

log.debug(f"log upload failed (swallowed, {len(batch)} entries dropped): {e}")

finally:

_FLUSH_GUARD.active = False

class UploadingLogHandler(logging.Handler):

def __init__(self, uploader: BatchedLogUploader, context: str = "socket-python-cli"):

super().__init__()

self._uploader = uploader

self._context = context

def emit(self, record: logging.LogRecord) -> None:

if getattr(_FLUSH_GUARD, "active", False):

return

try:

self._uploader.add({

"timestamp": _now_str(),

"level": logging.getLevelName(record.levelno),

"message": self.format(record),

"context": self._context,

})

except Exception:

self.handleError(record)

import logging

from typing import Optional

from .cli_client import CliClient

from .cli_run import finalize_cli_run, register_cli_run

from .log_uploader import BatchedLogUploader, UploadingLogHandler

class StreamingLogs:

def __init__(

self,

*,

client: CliClient,

cli_logger: logging.Logger,

sdk_logger: logging.Logger,

client_version: str,

upload_logs: Optional[bool],

enable_debug: bool,

):

self._client = client

self._loggers = (cli_logger, sdk_logger)

self._client_version = client_version

self._upload_logs = upload_logs

self._enable_debug = enable_debug

self._run_id: Optional[str] = None

self._report_run_id: Optional[str] = None

self._uploader: Optional[BatchedLogUploader] = None

self._upload_handler: Optional[UploadingLogHandler] = None

self._terminal_handler: Optional[logging.StreamHandler] = None

self._saved_levels: tuple = ()

self._saved_propagate: tuple = ()

def set_report_run_id(self, report_run_id: Optional[str]) -> None:

self._report_run_id = report_run_id

def __enter__(self) -> "StreamingLogs":

self._run_id = register_cli_run(

self._client,

client_version=self._client_version,

upload_logs=self._upload_logs,

)

cli_logger = self._loggers[0]

if not self._run_id:

cli_logger.debug("server log streaming not active for this run")

return self

self._uploader = BatchedLogUploader(self._client, self._run_id)

self._uploader.start()

self._upload_handler = UploadingLogHandler(self._uploader, context="socket-python-cli")

self._upload_handler.setFormatter(logging.Formatter("%(message)s"))

self._terminal_handler = logging.StreamHandler()

self._terminal_handler.setLevel(logging.DEBUG if self._enable_debug else logging.INFO)

self._terminal_handler.setFormatter(logging.Formatter("%(asctime)s: %(message)s"))

self._saved_levels = tuple(lg.level for lg in self._loggers)

self._saved_propagate = tuple(lg.propagate for lg in self._loggers)

for lg in self._loggers:

lg.setLevel(logging.DEBUG)

lg.propagate = False

lg.addHandler(self._terminal_handler)

lg.addHandler(self._upload_handler)

cli_logger.debug(f"server log streaming enabled (run_id={self._run_id})")

return self

def __exit__(self, exc_type, exc_val, exc_tb) -> bool:

if self._run_id is None:

return False

status = self._status_for_exit(exc_type, exc_val)

for lg in self._loggers:

lg.removeHandler(self._upload_handler)

self._uploader.stop()

finalize_cli_run(

self._client,

self._run_id,

status=status,

report_run_id=self._report_run_id,

)

for lg in self._loggers:

lg.removeHandler(self._terminal_handler)

for lg, level, propagate in zip(self._loggers, self._saved_levels, self._saved_propagate):

lg.setLevel(level)

lg.propagate = propagate

return False

@staticmethod

def _status_for_exit(exc_type, exc_val) -> str:

if exc_type is None:

return "success"

if issubclass(exc_type, KeyboardInterrupt):

return "cancelled"

# SystemExit with code 0 / None is a clean exit; non-zero codes signal failure.

if issubclass(exc_type, SystemExit) and not getattr(exc_val, "code", None):

return "success"

return "failure"