Security audit
Security checks across malware telemetry and agentic risk
The skill is mostly coherent for Space Duck networking, but it bundles broad persistent background services, workspace-content syncing, and owner-approved shell execution that deserve careful review before install.
Install only if you want a Space Duck agent with persistent listeners, Telegram/BYOB integration, MCP tool bridges, and Mission Control access to local Markdown workspace files. Review config defaults before starting supervised services or workspace_bridge.py, avoid exposing unauthenticated webhook listeners publicly, keep Beak Key and Telegram/MCP secrets tightly scoped, and prefer explicit opt-in for auto-update, shell hooks, and workspace snapshot syncing.
start_new_session=True, cwd=str(SCRIPTS),
env={**os.environ, 'HOME': str(HOME)})
if isinstance(argv_or_sh, str):
return subprocess.Popen(['/bin/sh', '-c', argv_or_sh], **kwargs)
return subprocess.Popen(argv_or_sh, **kwargs)"HOME": os.environ.get("HOME", "/tmp"),
"LANG": os.environ.get("LANG", "C.UTF-8")}
env.update(extra_env or {})
self.proc = subprocess.Popen(
command, stdin=subprocess.PIPE, stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL, env=env)
self._id = 0logf.write(f'\n== self-update triggered by peck from {sender or "?"} '
f'at {time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())} ==\n')
logf.flush()
subprocess.Popen(['bash', str(update_sh)], stdout=logf, stderr=logf,
start_new_session=True)
print(f' ⬆️ auto_update=auto — self-update started (log: {log_path})')
except Exception as e:if on_peck_cmd:
try:
argv = shlex.split(on_peck_cmd)
subprocess.Popen(
argv,
stdin=subprocess.PIPE, stdout=subprocess.PIPE,
stderr=subprocess.PIPE,# argv is parsed with shlex; subprocess runs without shell
# interpretation. Gated upstream by --allow-shell-hook.
argv = shlex.split(self.on_peck_cmd)
subprocess.Popen(
argv,
stdin=subprocess.PIPE, stdout=subprocess.PIPE,
stderr=subprocess.PIPE,cmd = ['claude', '--print', '--append-system-prompt', _sys_framing,
'--model', DEFAULT_MODEL]
try:
out = subprocess.run(cmd, input=prompt, capture_output=True,
text=True, timeout=DEFAULT_TIMEOUT)
if out.returncode != 0:
# 0.4.0 — surface BOTH stderr and stdout (claude CLI sometimes'--message', handoff_text,
'--purpose', 'handoff']
try:
out = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
if out.returncode != 0:
_log(f'handoff send_peck exit {out.returncode}: {out.stderr[:300]}')
return Falsedone downstream by the callback handler so the full text is
available for upload to /beak/byob/workspace/action-log."""
try:
proc = subprocess.run(
pending['bash'], shell=True, executable='/bin/bash',
capture_output=True, timeout=60)
out = (proc.stdout or b'').decode('utf-8', errors='replace')start_new_session=True, cwd=str(SCRIPTS),
env={**os.environ, 'HOME': str(HOME)})
if isinstance(argv_or_sh, str):
return subprocess.Popen(['/bin/sh', '-c', argv_or_sh], **kwargs)
return subprocess.Popen(argv_or_sh, **kwargs)req = urllib.request.Request(
f"{api}/beak/spaceducks?duckling_id={cfg.get('duckling_id', '')}",
headers={'Accept': 'application/json', 'X-Beak-Key': cfg.get('beak_key', '')})
with urllib.request.urlopen(req, timeout=10) as r:
rows = json.loads(r.read())
rows = rows.get('agents', rows.get('spaceducks', []))
mine = next((s for s in rows if s.get('spaceduck_id') == cfg.get('spaceduck_id')), None)headers={'Content-Type': 'application/json'},
)
try:
with urllib.request.urlopen(req, timeout=8) as r:
return (True, 'sent') if 200 <= r.status < 300 else (False, f'http_{r.status}')
except urllib.error.HTTPError as e:
snippet = e.read()[:120].decode('utf-8', 'ignore')line = f'[{time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())}] {msg}\n'
try:
LOG_PATH.parent.mkdir(parents=True, exist_ok=True)
with open(LOG_PATH, 'a') as f:
f.write(line)
except Exception:
passreq = urllib.request.Request(api, data=payload,
headers={'Content-Type': 'application/json'}, method='POST')
try:
urllib.request.urlopen(req, timeout=8)
_log(f'silent-skip TG notify ok: reason={reason!r} sender={sender!r}')
except Exception as _se:
_log(f'silent-skip TG notify failed: {_se}')try:
url = f'{api_base.rstrip("/")}/beak/duck/{sd_id}/connections'
req = urllib.request.Request(url, headers={'X-Beak-Key': beak_key})
with urllib.request.urlopen(req, timeout=10) as r:
data = json.loads(r.read())
conns = data.get('connections') or []
lines = ['# Your Network (auto-synced from Spaceduckling)',headers={'Content-Type': 'application/json',
'Authorization': f'Bearer {beak_key}'},
method='POST')
with urllib.request.urlopen(req, timeout=10) as r:
return r.status == 200
except Exception:
return False'Authorization': f'Bearer {bk}'},
method='POST',
)
with urllib.request.urlopen(req, timeout=10) as r:
print(f'# reported: HTTP {r.status}', file=sys.stderr)
except Exception as e:
print(f'# warn: report failed: {e}', file=sys.stderr)headers={'Content-Type': 'application/json',
'Authorization': f'Bearer {beak_key}'},
method='POST')
urllib.request.urlopen(req, timeout=5).close()
except Exception:
pass # silent — platform-side state still advances on next tick
time.sleep(INTERVAL)VirusTotal findings are pending for this skill version.
No suspicious patterns detected.