Security checks across malware telemetry and agentic risk
The skill mostly matches a BoTTube video SDK, but the bundle exposes live-looking credentials and includes automation scripts that can post, comment, vote, and modify a deployed server.
Install only after reviewing and removing or rotating the bundled credentials, and do not run the autonomous engagement, social orchestration, payout, GPU worker, webhook, or server patch scripts unless you intentionally operate this BoTTube deployment. Use your own scoped API key, keep TLS verification enabled, pin trusted service endpoints, and run media workers in an isolated low-privilege environment.
)
try:
result = subprocess.run(
cmd, shell=True, capture_output=True, text=True, timeout=30
)
if result.returncode != 0:cmd.extend(["-c:v", "libx264", "-preset", "fast", output_path])
print(f" Running: {' '.join(cmd)}")
result = subprocess.run(cmd, capture_output=True, timeout=300)
if result.returncode != 0:
return False, "", f"ffmpeg failed: {result.stderr.decode()[-200:]}""-c:a", "aac", "-b:a", "128k",
output_path
]
result = subprocess.run(cmd, capture_output=True, timeout=600)
if result.returncode != 0:
return False, "", f"Transcode failed: {result.stderr.decode()[-200:]}"str(output_path),
]
try:
result = subprocess.run(cmd, capture_output=True, text=True, timeout=60)
return result.returncode == 0 and output_path.exists()
except Exception:
return Falsedef _get_pixel_stats(frame_path: str) -> dict:
"""Use ffprobe to get basic pixel statistics from a frame."""
try:
result = subprocess.run(
[
"ffprobe", "-v", "error",
"-show_entries", "frame=width,height",Falls back to file-size heuristic if filter unavailable.
"""
try:
result = subprocess.run(
[
"ffprobe", "-v", "error",
"-show_entries","""
try:
# Use ffmpeg to compute mean and stdev of luminance
result = subprocess.run(
[
"ffprobe", "-v", "error",
"-show_entries",url = f"{BASE_URL}{endpoint}"
headers = {"X-API-Key": api_key, "Content-Type": "application/json"}
try:
r = requests.post(url, json=data, headers=headers, verify=VERIFY_SSL, timeout=15)
return r.status_code, r.json() if r.headers.get("content-type", "").startswith("application/json") else r.text
except Exception as e:
return 0, str(e)method="POST",
)
try:
resp = urllib.request.urlopen(req, timeout=10)
return jsonify({"ok": True, "status": resp.status})
except Exception as e:
return jsonify({"ok": False, "error": str(e)}), 502method="POST",
)
try:
with urllib.request.urlopen(req, timeout=10) as resp:
if 200 <= getattr(resp, "status", 200) < 300:
ok = True
break"""Download an image to WORK_DIR."""
filepath = WORK_DIR / filename
log.info(f"Downloading: {url[:80]}...")
r = requests.get(url, timeout=60, stream=True)
r.raise_for_status()
with open(filepath, "wb") as f:
for chunk in r.iter_content(8192):headers = {"X-API-Key": BOTTUBE_API_KEY}
try:
r = requests.get(
f"{BOTTUBE_URL}/api/videos",
params={"per_page": 20},
timeout=30,with open(file_path, 'rb') as f:
files = {'video': (os.path.basename(file_path), f, 'video/mp4')}
data = {'title': title, 'description': desc}
r = requests.post(url, headers=headers, files=files, data=data, timeout=30)
if 200 <= r.status_code < 300:
print(f"Uploaded {file_path}: {r.status_code} {r.text}")
else:"type": "output",
})
try:
with urllib.request.urlopen(
f"{COMFYUI_URL}/view?{params}", timeout=30
) as resp:
data = resp.read())
try:
result = subprocess.run(
cmd, shell=True, capture_output=True, text=True, timeout=30
)
if result.returncode != 0:external_id,
headers={"Authorization": f"Token {REPLICATE_API_TOKEN}"},
)
with urllib.request.urlopen(req, timeout=10) as resp:
data = json.loads(resp.read())
status = data.get("status", "starting")
if status == "succeeded":external_id,
headers={"Authorization": f"Token {REPLICATE_API_TOKEN}"},
)
with urllib.request.urlopen(req, timeout=30) as resp:
data = json.loads(resp.read())
output_url = data.get("output", "")if not _transcode_to_mp4(result_path, final_path):
# If transcode fails but source is already mp4, try direct copy
if result_path.suffix == ".mp4":
shutil.copy2(result_path, final_path)
else:
log.warning("Transcode failed for provider %s", provider_name)
final_path.unlink(missing_ok=True)headers={"Content-Type": "application/json"},
method="POST",
)
with urllib.request.urlopen(req, timeout=15) as resp:
result = json.loads(resp.read())
prompt_id = result.get("prompt_id")
if not prompt_id:while time.time() < deadline:
time.sleep(3)
try:
with urllib.request.urlopen(f"{COMFYUI_URL}/history/{prompt_id}", timeout=10) as resp:
history = json.loads(resp.read())
if prompt_id in history:
entry = history[prompt_id]for _ in range(40): # Up to 200 seconds
time.sleep(5)
req2 = urllib.request.Request(poll_url, headers={"Authorization": f"Token {REPLICATE_API_TOKEN}"})
with urllib.request.urlopen(req2, timeout=10) as resp2:
status = json.loads(resp2.read())
if status.get("status") == "succeeded":
break"type": "output",
})
try:
with urllib.request.urlopen(
f"{COMFYUI_URL}/view?{params}", timeout=30
) as resp:
data = resp.read()data=payload,
headers={"Content-Type": "application/json"},
)
with urllib.request.urlopen(req, timeout=VISION_TIMEOUT) as resp:
data = json.loads(resp.read())
response_text = data.get("response", "")
except Exception as e2:data=payload,
headers={"Content-Type": "application/json"},
)
with urllib.request.urlopen(req, timeout=VISION_TIMEOUT) as resp:
result = json.loads(resp.read())
quality_score = result.get("quality_score", 5)data=img_data,
headers={"Content-Type": "application/octet-stream"},
)
with urllib.request.urlopen(req, timeout=10) as resp:
result = json.loads(resp.read())
detection = result.get("detection", {})66/66 vendors flagged this skill as clean.