Something went wrong. Try again.
small gleam coding and persistent agent daemon with a detachable cli and a canvas ui forked from okami.mom/albedo
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100"""Refuse output containing provided secrets or recognizable credential formats."""
from __future__ import annotations
import base64import htmlimport jsonimport osfrom pathlib import Pathimport reimport tempfilefrom urllib.parse import unquote
SECRET_NAME = re.compile( r"token|secret|password|api.?key|credential|authorization", re.I)KEY_PATTERNS = ( re.compile(r"sk-(?:proj-|or-v1-)?[A-Za-z0-9_-]{20,}"), re.compile(r"gh[pousr]_[A-Za-z0-9]{20,}"), re.compile(r"github_pat_[A-Za-z0-9_]{20,}"), re.compile(r"(?:AKIA|ASIA)[0-9A-Z]{16}"), re.compile(r"eyJ[A-Za-z0-9_-]{15,}\.[A-Za-z0-9_-]{15,}\.[A-Za-z0-9_-]{10,}"), re.compile(r"-----BEGIN (?:RSA |EC |OPENSSH )?PRIVATE KEY-----"),)
class SecretScanner: def __init__(self, extra=()): self.values = { value for key, value in os.environ.items() if SECRET_NAME.search(key) and len(value) >= 8 } self.values.update(value for value in extra if value and len(value) >= 8)
def check(self, content, *, label="output"): def check_text(text): for candidate in (text, html.unescape(text), unquote(text)): if any(value in candidate for value in self.values): raise ValueError( f"secret scan rejected {label}: provided credential value" ) if any(pattern.search(candidate) for pattern in KEY_PATTERNS): raise ValueError( f"secret scan rejected {label}: credential-shaped string" )
def walk(value): if isinstance(value, dict): if "body" in value and "status" in value: body = base64.b64decode(value["body"]).decode() walk(body) for item in value.values(): walk(item) elif isinstance(value, list): for item in value: walk(item) elif isinstance(value, bytes): walk(value.decode("utf-8", errors="replace")) elif isinstance(value, str): check_text(value) # Credentials may span SSE deltas. Scan reconstructed model text # and tool arguments as well as each original response byte. fragments = [] for line in value.splitlines(): if line.startswith("data: ") and line != "data: [DONE]": event = json.loads(line[6:]) for choice in event.get("choices", []): delta = choice.get("delta", {}) fragments.extend( v for v in ( delta.get("content"), delta.get("reasoning_content"), ) if isinstance(v, str) ) for tool in delta.get("tool_calls", []): fragments.append( tool.get("function", {}).get("arguments", "") ) if fragments: check_text("".join(fragments))
walk(content)
def write_json(self, path, data): path = Path(path) self.check(data, label=str(path)) text = json.dumps(data, indent=2) + "\n" self.check(text, label=str(path)) path.parent.mkdir(parents=True, exist_ok=True) descriptor, temporary = tempfile.mkstemp(prefix=".perf-", dir=path.parent) try: with os.fdopen(descriptor, "w") as handle: handle.write(text) os.replace(temporary, path) finally: Path(temporary).unlink(missing_ok=True)