Is headroomlabs-ai/headroom safe?

compile(code, "<headroom-compressed>", "exec")

return ExportWrapper(core).eval()

"Bearer sk-a…[redacted, 49 chars]",

"Bearer sk-a…[redacted, 40 chars]",

$cmdWrapper = ([string][char]64) + "echo off`r`npowershell -NoLogo -NoProfile -ExecutionPolicy Bypass -File ""%~dp0headroom.ps1"" %*`r`n"

import { spawn } from "node:child_process";

import { spawnSync } from "node:child_process";

const child = spawn(spec.command, spec.args, {

const result = spawnSync("pyenv", ["which", "headroom"], {

const result = spawnSync(command, args, {

const result = spawnSync("npm", ["prefix", "-g"], {

import { spawnSync } from "node:child_process";

const result = spawnSync(command, args, {

import { spawnSync } from "node:child_process";

const result = spawnSync("tar", ["-xzf", tarballPath, "-C", workdir], {

return spawnSync("cmd.exe", ["/d", "/s", "/c", "npm.cmd", ...args], {

return spawnSync("npm", args, {

const result = spawnSync(process.execPath, ["--input-type=module", "-e", smoke], {

return _sp.run(*args, **kwargs)

return _sp.Popen(*args, **kwargs)

result = subprocess.run(command)

completed = subprocess.run(cmd, check=False)

result = subprocess.run(argv)  # noqa: S603

subprocess.Popen(handoff_argv)  # noqa: S603 - fixed allowlist plus helper argv

proc = subprocess.Popen(cmd, **popen_kwargs)

proc = subprocess.Popen(cmd, **popen_kwargs)

subprocess.run(
                ["taskkill", "/F", "/T", "/PID", str(proc.pid)],
                capture_output=True,
                timeout=10,
                check=False,
            )

proc = subprocess.Popen(
            [
                "uvx",
                # PyPI (prebuilt wheels), not the git source that fails to build
                # under proot-based filesystems (#2871).
                "--from", …

subprocess.run(
                ["taskkill", "/F", "/T", "/PID", str(pid)],
                capture_output=True,
                timeout=10,
                check=False,
            )

result = subprocess.run([binary, *args], env=env)

result = subprocess.run([claude_bin, *claude_args], env=env)

proc = subprocess.Popen(
            [sys.executable, "-m", "headroom.proxy.server", "--port", str(port)],
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
        )

subprocess.run(command, check=True)

subprocess.run(command, check=True)

proc = subprocess.Popen(command, env=env, stdout=log_file, stderr=log_file)

proc = subprocess.Popen(command, **kwargs)

subprocess.run(docker_cmd, check=True)

subprocess.Popen(command, **popen_kwargs)

subprocess.run(
                ["schtasks", "/Create", "/TN", name, "/XML", tmp.name, "/F"],
                check=True,
            )

query = subprocess.run(
                ["schtasks", "/Query", "/TN", name, "/XML"],
                check=True,
                capture_output=True,
            )

subprocess.run(["systemctl", *flags, "daemon-reload"], check=True)

subprocess.run(["systemctl", *flags, "enable", manifest.service_name], check=True)

subprocess.run(["systemctl", *flags, "restart", manifest.service_name], check=True)

CONTAINER_BIND_HOST = "0.0.0.0"  # noqa: S104 — container-internal bind, published only on 127.0.0.1

body = signature.split("::", 1)[-1]

With the default ``--host 0.0.0.0`` Docker bind, leaving it open would

``--host 0.0.0.0`` Docker bind, an unauthenticated POST from any

await fs.readFile(path.join(rootDir, "package.json"), "utf8"),

await fs.writeFile(

with open(f"/proc/{pid}/stat", "rb") as fh:

with open(path, encoding="utf-8") as f:

with open(path, encoding="utf-8") as f:

with open(path, encoding="utf-8") as f:

rows = [json.loads(x) for x in open(os.environ["HEADROOM_CACHE_TTL_OBS_PATH"])]

for line_number, line in enumerate(capture_path.read_text(encoding="utf-8").splitlines(), 1):

with open(SHARED_STATS_FILE) as f:

content = safe_decode_for_logging(path.read_bytes())

payload = json.loads(path.read_text(encoding="utf-8"))

payload = json.loads(path.read_text(encoding="utf-8")) if path.exists() else {}

payload = json.loads(path.read_text(encoding="utf-8"))

payload = json.loads(path.read_text(encoding="utf-8"))

payload = json.loads(settings_path.read_text(encoding="utf-8"))

text = candidate.read_text(encoding="utf-8", errors="replace")

payload = json.loads(savings_file.read_text(encoding="utf-8"))

content = path.read_text(encoding="utf-8").strip()

content = path.read_text(encoding="utf-8") if path.exists() else ""

content = path.read_text(encoding="utf-8") if path.exists() else ""

payload = json.loads(path.read_text(encoding="utf-8")) if path.exists() else None

text = path.read_text(encoding="utf-8", errors="replace") if path.exists() else ""

record = _json.loads((ws / _VERBOSITY_PIN_FILE).read_text(encoding="utf-8"))

with open(MCP_CONFIG_PATH, encoding="utf-8") as f:

with open(path) as f:

content = fsutil.read_text(file_path)

payload = json.loads(marker.read_text(encoding="utf-8"))

args.output.write_text(json.dumps(payload, ensure_ascii=False) + "\n", encoding="utf-8")

shutil.copyfileobj(src, out)

shutil.copyfileobj(gz, out)

shutil.copy2(archive, dest)

shutil.copyfileobj(extracted, out)

shutil.copyfileobj(src, out)

shutil.move(str(staging), tmp_final)

with open(tmp, "w", encoding="utf-8") as f:

with open(path, "a", encoding="utf-8") as f:

with open(os.environ["HEADROOM_CACHE_TTL_LEARNED_PATH"], "w") as f:

with open(SHARED_STATS_FILE, "a") as f:

with open(SHARED_STATS_FILE, "w") as f:

(logs_dir / "proxy.log").write_text("\n".join(perf_lines) + "\n", encoding="utf-8")

eval_path.write_text(
        json.dumps(
            {
                "totals": {
                    "cases": 3,
                    "passed": 3,
                    "accuracy_rate": 1.0,
                    "tokens_original": 3000, …

markdown_output.write_text(markdown, encoding="utf-8")

json_output.write_text(json.dumps(diff.to_dict(), indent=2), encoding="utf-8")

json_output.write_text(json_module.dumps(report.to_dict(), indent=2), encoding="utf-8")

json_output.write_text(json_module.dumps(report.to_dict(), indent=2), encoding="utf-8")

path.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8")

path.write_text(content, encoding="utf-8")

path.write_text(content, encoding="utf-8")

with open(os.devnull, "w", encoding="utf-8") as devnull:

path.write_text(_json.dumps(record), encoding="utf-8", newline="\n")

with open(MCP_CONFIG_PATH, "w", encoding="utf-8") as f:

fsutil.write_text(output_path, json_output)

async fetch(request, env, ctx) {

fetch(env.METRICS_OTLP_URL, {

const resp = await fetch(url2, {

const res = await $.http.fetch(`${live.url}/stats?cached=1`)

const response = await fetch(`${origin}${path}`, {

const resp = await fetch(url, {

const resp = await fetch(url, {

response = await fetch(url, {

import urllib.request

req = urllib.request.Request(final_url, headers={"User-Agent": "headroom-binaries/1"})

resp = urllib.request.urlopen(req, timeout=60, context=ctx)  # noqa: S310 (https)

resp = urllib.request.urlopen(req, timeout=60)  # noqa: S310 (https)

url_path = urllib.parse.urlparse(url).path

import httpx

import httpx

self._http_client = httpx.AsyncClient(timeout=15.0)

async with httpx.AsyncClient(timeout=5.0) as client:

self._http_client = httpx.AsyncClient(timeout=15.0)

import urllib.request

with urllib.request.urlopen(url, timeout=2) as response:

import urllib.request

request = urllib.request.Request(
        f"http://127.0.0.1:{resolved_port}/admin/runtime-env",
        data=_json.dumps(overrides).encode("utf-8"),
        method="POST",
        headers={"Content-Type": "application/json"},
    )

with urllib.request.urlopen(request, timeout=2) as response:

import httpx

response = httpx.get(f"{proxy_url}/health", timeout=2.0)

import urllib.request

with urllib.request.urlopen(url, timeout=2) as response:

parsed = urllib.parse.urlparse(raw)

import urllib.request

request = urllib.request.Request(
        f"http://127.0.0.1:{port}/admin/runtime-env",
        data=json.dumps(payload).encode("utf-8"),
        method="POST",
        headers={"Content-Type": "application/json"},
    )

with urllib.request.urlopen(request, timeout=2) as response:

parsed = urllib.parse.urlsplit(base_url)

from urllib import error as urllib_error

return _id_token_carries_chatgpt_account(tokens.get("id_token"))

"""Whether an ``id_token`` carries the ChatGPT account claim (#3206).

the ``id_token`` claims. Those configs then read as API-key mode, so

of it. An API-key user has no ChatGPT id_token, so this cannot resurrect

"id_token",

"tools": {

{

@tool

@tool

@tool

@tool

@tool

@server.list_tools()

@server.call_tool()

api.registerTool((ctx: any) => {

Tool(
                    name=COMPRESS_TOOL_NAME,

Tool(
        name="memory_search",