egress.py

Go to the documentation of this file.

"""Plain-text outbound scanner derived from OCT's secret-scanner heuristics.
 
Provenance: the pattern vocabulary and Shannon-entropy formula are ported from
``oct/oct/tools/secret_scanner.py``. OCT's scanner itself is intentionally not
imported: it walks Python AST nodes and cannot certify ordinary message prose.
This module scans content tokens and well-known credential shapes instead.
 
Dependencies
------------
- oc_diagnostics (_dbg structured logging of text length and finding count per scan, and of each block by finding count only)
"""
from __future__ import annotations
 
import math
import re
from collections import Counter
from dataclasses import dataclass
 
from oc_diagnostics import _dbg
 
_MIN_ENTROPY_LENGTH = 24
_ENTROPY_THRESHOLD = 4.5
 
_GITHUB_SHAPE = re.compile("gh" + "[pousr]_" + "[A-Za-z0-9]" + "{20,}")
_AWS_SHAPE = re.compile("AKIA" + "[0-9A-Z]" + "{16}")
_OPENAI_SHAPE = re.compile("sk-" + "[A-Za-z0-9]" + "{20,}")
_TELEGRAM_SHAPE = re.compile("[0-9]" + "{6,12}:" + "[A-Za-z0-9_-]" + "{30,}")
_TOKEN_SHAPE = re.compile("[A-Za-z0-9_" + "+/=-]" + "{24,}")
 
_LABELS = (
    "api[ _-]?key",
    "access[ _-]?token",
    "auth[ _-]?token",
    "client[ _-]?secret",
    "password",
    "private[ _-]?key",
    "refresh[ _-]?token",
    "secret",
    "token",
)
_NAMED_VALUE_SHAPE = re.compile(
    "\\b(?:"
    + "|".join(_LABELS)
    + ")\\b"
    + "\\s*[:=]\\s*"
    + "[\"']?"
    + "([A-Za-z0-9_"
    + "+/=.:-]{8,})",
    re.IGNORECASE,
)
 
THREE_OPTIONS_ESCALATION = (
    "Outbound message blocked: a possible credential was detected and nothing was sent.\n"
    "1. Remove or redact the sensitive value, then retry.\n"
    "2. Ask the operator to review a possible false positive through the scoped waiver workflow.\n"
    "3. Cancel the send and keep the message local."
)
 
 
@dataclass(frozen=True)
class EgressFinding:
    """A non-sensitive description of one blocked span."""
 
    kind: str
    start: int
    end: int
    entropy: float | None = None
 
 
class EgressBlockedError(PermissionError):
    """Outbound content failed the mandatory edge scan."""
 
    def __init__(self, findings: tuple[EgressFinding, ...]) -> None:
        self.findings = findings
        kinds = ", ".join(sorted({finding.kind for finding in findings}))
        super().__init__(f"{THREE_OPTIONS_ESCALATION}\nFinding classes: {kinds}.")
 
 
def shannon_entropy(value: str) -> float:
    """Return Shannon entropy in bits per character."""
    if not value:
        return 0.0
    counts = Counter(value)
    length = len(value)
    return -sum(
        (count / length) * math.log2(count / length)
        for count in counts.values()
    )
 
 
def scan_text(text: str) -> tuple[EgressFinding, ...]:
    """Return content-oriented findings without returning matched content."""
    if not isinstance(text, str):
        raise TypeError("outbound content must be text")
    findings: list[EgressFinding] = []
    for kind, shape in (
        ("github-token-shape", _GITHUB_SHAPE),
        ("aws-key-shape", _AWS_SHAPE),
        ("openai-key-shape", _OPENAI_SHAPE),
        ("telegram-token-shape", _TELEGRAM_SHAPE),
    ):
        findings.extend(
            EgressFinding(kind=kind, start=match.start(), end=match.end())
            for match in shape.finditer(text)
        )
    findings.extend(
        EgressFinding(
            kind="named-sensitive-value",
            start=match.start(1),
            end=match.end(1),
        )
        for match in _NAMED_VALUE_SHAPE.finditer(text)
    )
 
    covered = {(finding.start, finding.end) for finding in findings}
    for match in _TOKEN_SHAPE.finditer(text):
        span = (match.start(), match.end())
        if span in covered or any(
            start <= span[0] and span[1] <= end for start, end in covered
        ):
            continue
        candidate = match.group(0)
        if len(candidate) < _MIN_ENTROPY_LENGTH:
            continue
        entropy = shannon_entropy(candidate)
        if entropy >= _ENTROPY_THRESHOLD:
            findings.append(
                EgressFinding(
                    kind="high-entropy-token",
                    start=span[0],
                    end=span[1],
                    entropy=entropy,
                )
            )
 
    unique = {
        (finding.kind, finding.start, finding.end): finding for finding in findings
    }
    result = tuple(unique[key] for key in sorted(unique))
    _dbg(
        "OC_CONNECT.EGRESS",
        "scan_text",
        f"scanned outbound text length={len(text)} findings={len(result)}",
        3,
    )
    return result
 
 
def require_safe(text: str) -> None:
    """Return only for clean text; otherwise block with three operator options."""
    findings = scan_text(text)
    if findings:
        _dbg(
            "OC_CONNECT.EGRESS",
            "require_safe",
            # OCT-LINT: disable=no_hardcoded_secrets reason="Security diagnostic prose reports only a finding count; no credential content is present"
            f"SECURITY_ERROR: blocked outbound text findings={len(findings)}",
            1,
            override=True,
        )
        raise EgressBlockedError(findings)
 
 
__all__ = [
    "EgressBlockedError",
    "EgressFinding",
    "THREE_OPTIONS_ESCALATION",
    "require_safe",
    "scan_text",
    "shannon_entropy",
]