Security checks across malware telemetry and agentic risk
This appears to be a real Space Duck client, but it enables persistent listeners, workspace file access, auto-replies, and remotely triggered shell actions that users should review carefully before installing.
Install only if you intend to give Space Duck a persistent local bridge with access to your agent workspace and Beak Key. Bind listeners behind trusted TLS/firewalling, avoid exposing the unauthenticated peck listener publicly, disable or restrict auto-reply and remembered approvals unless needed, and review any Telegram or Mission Control action before allowing shell execution.
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', '--permission-mode', 'bypassPermissions',
'--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')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')url, method='GET',
headers={'X-Beak-Key': beak_key, 'X-Spaceduck-ID': sdid},
)
with urllib.request.urlopen(req, timeout=10) as r:
data = json.loads(r.read())
content = data.get('content', '')
(files_dir / fn).write_text(content)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:
passtry:
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)64/64 vendors flagged this skill as clean.