v13.0 — The Hardened & Secure Update
-3DDC84?logo=android&logoColor=white){kind=icon}
-blueviolet){kind=icon}
zM SOS GUARD is a sovereign smartphone “Black Box” engineered for critical and hostile environments.
It transforms any Android device into an indestructible forensic recorder that continuously captures high-integrity audio/video + GPS streams, commits every frame to local flash first, and delivers evidence over encrypted channels even when the network is intermittent, censored, or hostile.
Unlike commercial surveillance apps bloated with advertising SDKs, analytics, and heavy frameworks, zM SOS GUARD is built as a pure-vanilla, zero-dependency system:
- Client: Pure Kotlin + native Android SDK + AndroidX Security Crypto only.
- Server: Pure Python 3 standard library (zero
pip install). - Attack surface: minimal.
- Cold-start: near-instant.
- Survival priority: local disk always wins over the network.
The system is designed for personal protection, emergency documentation, technical research into mobile OS resilience, and legitimate forensic evidence preservation.
- Client: Pure Vanilla Kotlin/XML. No Jetpack Compose, no OkHttp, no Retrofit, no Glide, no Firebase. Only the official AndroidX Security Crypto artifact for hardware-backed key storage.
- Server: Pure Python 3 standard library. No Flask, FastAPI, Django, or any third-party package.
- Result: sub-1024 KB APK, near-zero cold-start latency, and a dramatically reduced attack surface.
Video recording and network transmission run on independent threads.
A network outage, DNS failure, or TLS handshake timeout never stalls or drops frames from the camera pipeline.
Every completed chunk is written to local flash (zMPanicRec/) before any uplink attempt is made.
The network is a best-effort delivery layer; the disk is the source of truth.
Successfully uploaded files are renamed to *.synced.mp4.
They are never deleted by the application. The operator retains full control over evidence retention policy.
Remote endpoints are queried sequentially (semicolon-delimited list).
The first HTTP 2xx response ends the attempt for that chunk, conserving battery and mobile data.
┌─────────────────────────────────────────────────────────────────────────────┐
│ ANDROID CLIENT (PanicService) │
│ │
│ ┌──────────────┐ ┌──────────────────────┐ ┌─────────────────────────┐ │
│ │ Ghost Engine │ │ Dual-Mode Camera │ │ Local Flash Storage │ │
│ │ (Stealth) │ │ Engine │ │ zMPanicRec/ │ │
│ │ - Mute audio │ │ Mode A: Gapless │ │ SOS_*.mp4 │ │
│ │ - SECRET notif│ │ setNextOutputFile │ │ (never deleted) │ │
│ │ - Icon alias │ │ Mode B: Timer rotate │ │ *.synced.mp4 flags │ │
│ │ - moveTaskTo │ │ Hardware Watchdog │ └────────────┬────────────┘ │
│ │ Back │ │ +5s 803 timeout │ │ │
│ └──────────────┘ └──────────┬───────────┘ │ │
│ │ │ │
│ ▼ ▼ │
│ ┌─────────────────────┐ ┌───────────────────────┐ │
│ │ GPS / LocationManager│ │ Store-and-Forward │ │
│ │ live coordinates │ │ chronological queue │ │
│ └──────────┬──────────┘ │ multi-server failover │ │
│ │ └───────────┬───────────┘ │
│ │ │ │
│ └─────────────┬─────────────┘ │
│ │ │
│ ▼ │
│ Encrypted HTTPS / TLS │
│ Authorization: Bearer <pwd> │
│ File-Name + GPS headers │
└──────────────────────────────────────────────┼─────────────────────────────┘
│
│ Strict TLS only
▼
┌─────────────────────────────────────────────────────────────────────────────┐
│ HARDENED SERVER NODE (SERVER.py) │
│ │
│ ┌──────────────────┐ ┌──────────────────┐ ┌──────────────────────────┐ │
│ │ PBKDF2-HMAC-SHA256│ │ In-Memory Rate │ │ Anti-Collision Storage │ │
│ │ 250 000 iterations│ │ Limiting │ │ SOS_{ts}_{ip}_{uid}.mp4 │ │
│ │ 32-byte salt │ │ 5 fails → 1h ban │ │ + mirrored .meta.txt │ │
│ │ hmac.compare_digest│ └──────────────────┘ │ GPS + Google Maps URL │ │
│ └──────────────────┘ └──────────────────────────┘ │
│ │
│ Blind Drop-Box: POST /upload ONLY · No GET · No DELETE · No listing │
│ Auto-provisioned RSA-2048 self-signed certificate · Supervisor watchdog │
└─────────────────────────────────────────────────────────────────────────────┘
Mode A — Flagship Gapless Chunking (Android 8+)
setMaxFileSize( (videoBitrate + audioBitrate) / 8 × rotationSeconds )
→ MEDIA_RECORDER_INFO_MAX_FILESIZE_APPROACHING (802)
→ mediaRecorder.setNextOutputFile(nextFile)
→ MEDIA_RECORDER_INFO_NEXT_OUTPUT_FILE_STARTED (803)
→ seamless hardware-level hand-off, zero lost frames
Mode B — Legacy Fail-Safe Timer Rotation
For older or fragmented camera HALs that never emit event 803:
scheduleLegacyRotation() → rotateProcess()
mediaRecorder.stop() → reset() → startMediaRecorder()
Hardware Watchdog
A background timer armed to activeRotation + 5 seconds after every successful gapless transition.
If the 803 event is never received, the watchdog forces isGaplessSupported = false and permanently switches the process to Mode B for the remainder of the session, preventing silent lock-ups caused by buggy OEM HALs.
| Problem | Solution |
|---|---|
| Samsung Exynos Error -19 (preview/video size mismatch) | getBestCommonSize() — intersects supported preview sizes with supported video sizes and selects the highest common resolution. Preview size must equal video size. |
| Silent-mode muting & waterproof membrane dampening (Doogee, Blackview, etc.) | Forced AudioSource.MIC + 30 FPS. Bypasses silent-mode audio suppression and compensates for acoustic membrane attenuation. |
| Surface lifecycle crashes / duplicate attachments | Dummy SurfaceTexture(0) for background recording + careful weak-reference sharing of the visible TextureView surface only when the UI is present. |
MasterKey.Builder(context).setKeyScheme(MasterKey.KeyScheme.AES256_GCM)- Secrets stored exclusively in
EncryptedSharedPreferences(AES-256-SIV key encryption + AES-256-GCM value encryption). - Mandatory first-launch password dialog (1–512 characters). The password is never written in clear text.
When the EXTRA_HIDDEN flag is set:
AudioManagerzeroesSTREAM_SYSTEM,STREAM_RING,STREAM_NOTIFICATION.- Camera shutter sound disabled via
camera.enableShutterSound(false). - Notification channel switches to
VISIBILITY_SECRETwith iconstat_notify_syncand texts that mimic a normal cloud sync process (sys_integritychannel). - All Toasts and vibration are suppressed.
- Instant
moveTaskToBack(true). - Dynamic launcher icon disguise via
ActivityAlias(optional build-time configuration).
Version 15.0 introduces a new Anti-Stalking protection system designed to provide a delayed, multi-stage safety workflow before automatically activating Ghost Mode.
- Added a 30-minute countdown before the protection sequence proceeds.
- The extended countdown helps reduce accidental activations.
- The protection timer operates independently from the visible application interface.
Once the 30-minute countdown expires:
- A 15-second emergency alert is triggered.
- The alert uses a high-priority audio mechanism designed to remain audible even when the device is in silent mode, where supported by Android/OEM audio policies.
- The warning provides a final opportunity to intervene.
After the 15-second warning:
- A final 30-second confirmation window begins.
- The user can cancel the protection sequence during this period.
- If no valid confirmation/cancellation is received, the protection workflow proceeds automatically.
If the complete sequence reaches its activation point, the application automatically transitions into Ghost Mode.
The complete workflow is:
30-minute countdown
↓
15-second emergency alert
↓
30-second final confirmation
↓
Ghost Mode
v15.0 adds protection against manipulation of the Android system clock being used to interfere with the Anti-Stalking countdown.
The protection system does not blindly rely on the displayed wall-clock time.
Added protections:
- Detection of unexpected system-clock changes.
- Detection of abnormal elapsed-time behavior.
- Protection against simple forward/backward clock manipulation.
- Additional consistency checks during critical protection-state transitions.
- Use of monotonic/elapsed timing where appropriate.
Changing the Android date or time should therefore no longer provide a simple way to bypass the protection countdown.
Version 15.0 introduces TOFU (Trust On First Use) for server identity establishment.
The first trusted server identity is recorded and subsequently used as a reference for future connections.
TOFU provides:
- Initial server identity establishment.
- Persistent trust of the previously observed identity.
- Detection of unexpected server identity changes.
- Additional protection against silently switching to an untrusted server identity.
TOFU adds a dedicated trust layer to the existing encrypted communication architecture.
v15.0 introduces Elliptic Curve Cryptography (ECC) for asymmetric cryptographic operations.
The new architecture uses public/private key pairs to establish and verify cryptographic identity without transmitting private keys.
Added:
- Elliptic-curve public/private key pairs.
- Public-key based identity verification.
- Private-key protection on the originating device/system.
- Public-key trust material for identity verification.
- Integration between ECC identity and the new TOFU trust model.
The ECC layer provides a dedicated asymmetric cryptographic foundation for the new trust architecture.
The v15.0 security model now adds a cryptographic identity layer on top of the existing security mechanisms:
v15.0 Security Layer
│
┌───────┴───────┐
│ │
TOFU ECC
│ │
└───────┬───────┘
│
Server Identity
│
Trusted Endpoint
This is an additional security layer and does not replace the security mechanisms introduced in previous releases.
- 🚨 Added Anti-Stalking Protection System
- ⏱️ Added 30-minute protection countdown
- 🔊 Added 15-second high-priority emergency alert
- ⏳ Added 30-second final confirmation window
- 👻 Added automatic Ghost Mode activation
- 🕐 Added Android time-manipulation protection
- 🛡️ Improved countdown integrity
- 🔐 Added TOFU (Trust On First Use)
- 🔑 Added ECC-based asymmetric cryptography
- 🔒 Added cryptographic server identity verification
- 🔗 Integrated TOFU + ECC into the new trust architecture
🚀 v15.0
Focus: Anti-Stalking Protection · Ghost Mode · Time Integrity · TOFU · ECC Asymmetric Cryptography
Status: Stable
- Continuous local recording during network cuts.
- Chronological backlog flush (files sorted by
lastModified()) upon 4G/Wi-Fi reconnect. ConnectivityManager.NetworkCallback+bindProcessToNetworkfor process-level network binding.- Multi-server priority failover: semicolon-delimited endpoints (
IP1:PORT;IP2:PORTor fullhttps://…URLs). First HTTP 2xx wins. - Service returns
START_STICKY. PARTIAL_WAKE_LOCK(long timeout) +WIFI_MODE_FULL_HIGH_PERFprevent CPU and Wi-Fi radio from entering Doze/IDLE.
- 62+ native languages managed exclusively through Android resource qualifiers (
res/values-*/strings.xml). - Live GPS coordinate and status Toasts restored when
!activeHiddenMode.
- Neon electrical-style UI.
- Runtime permission orchestration (Camera, Microphone, Location, Notifications, Foreground Service types).
- First-launch mandatory password setup dialog.
- Live preview via
TextureViewsurface sharing with the service. - Configuration of server endpoints, rotation interval, front/rear camera, and stealth mode.
Zero pip dependencies. Pure Python 3 standard library.
Copy the entire block below into SERVER.py (or server.py) and run with python3 SERVER.py.
import http.server
import os
import time
import shutil
import sys
import random
import traceback
import socket
import threading
import uuid
import ssl
import subprocess
import signal
import re
import select
import getpass
import hashlib
import hmac
import tempfile
from http.server import HTTPServer
from socketserver import ThreadingMixIn
PORT = 9999
SAVE_DIR = "zmpanic_recordings"
MAX_FILE_SIZE = 100 * 1024 * 1024 # 100 MB
SOCKET_TIMEOUT = 10.0 # Anti-Slowloris timeout per raw read
HEADER_TIMEOUT = 30.0 # Max time allowed to complete TLS + HTTP headers
MAX_UPLOAD_TIME = 300.0 # Wall-clock max duration per upload
MIN_BYTE_RATE = 10 * 1024 # 10 KB/s minimum throughput required
MAX_CONCURRENT = 12 # Strict concurrency cap (no thread spawn beyond this)
MIN_FREE_DISK_BYTES = 500 * 1024 * 1024 # 500 MB safety buffer
CERT_FILE = ".cert.pem"
KEY_FILE = ".key.pem"
PASS_FILE = ".srvpass.txt"
# ===== RATE-LIMITING & INTERNAL BAN SYSTEM =====
MAX_FAILED_ATTEMPTS = 5
BAN_DURATION = 3600
failed_attempts = {}
banned_ips = {}
ban_lock = threading.Lock()
# ===== CONCURRENCY SEMAPHORE & SOCKET REGISTRY =====
concurrency_semaphore = threading.BoundedSemaphore(MAX_CONCURRENT)
active_sockets = {}
active_sockets_lock = threading.Lock()
# ===== TERMINAL FX =====
RESET = "\033[0m"
BOLD = "\033[1m"
DIM = "\033[2m"
CYAN = "\033[96m"
BLUE = "\033[94m"
GREEN = "\033[92m"
RED = "\033[91m"
YELLOW = "\033[93m"
WHITE = "\033[97m"
MAGENTA = "\033[95m"
CLEAR = "\033[2J\033[H"
start_time = time.time()
files_received = 0
total_bytes = 0
log_buffer = []
shutdown_flag = False
saved_password_hash = None
saved_salt = None
saved_online_verifier = None
last_redraw_time = 0.0
def ts():
return time.strftime("%Y-%m-%d %H:%M:%S")
def term_size():
try:
s = shutil.get_terminal_size()
return s.columns, s.lines
except Exception:
return 120, 40
def center_block(lines):
w, h = term_size()
block_h = len(lines)
top_padding = max(0, (h - block_h) // 2)
print(CLEAR, end="")
print("\n" * top_padding, end="")
for line in lines:
pad = max(0, (w - len(strip_ansi(line))) // 2)
print(" " * pad + line)
def strip_ansi(s):
return re.sub(r'\x1b\[[0-9;]*m', '', s)
def hr(char="═"):
w, _ = term_size()
return char * w
def box(title, lines):
w, _ = term_size()
inner = w - 4
print(CYAN + "╔" + "═" * (w - 2) + "╗" + RESET)
t = " " + title + " "
print(CYAN + "║" + BOLD + t + " " * (w - 2 - len(t)) + CYAN + "║" + RESET)
print(CYAN + "╠" + "═" * (w - 2) + "╣" + RESET)
for l in lines:
l2 = str(l)[:inner]
print(CYAN + "║ " + RESET + l2 + " " * (inner - len(strip_ansi(l2))) + CYAN + " ║" + RESET)
print(CYAN + "╚" + "═" * (w - 2) + "╝" + RESET)
def draw_header():
try:
uptime = int(time.time() - start_time)
w, _ = term_size()
print(CLEAR, end="")
print(MAGENTA + hr("═") + RESET)
print(MAGENTA + BOLD + " ZMPANIC :: SECURE EVIDENCE INGESTION NODE (HARDENED)".ljust(w) + RESET)
print(MAGENTA + hr("─") + RESET)
print(GREEN + f" LISTEN : 0.0.0.0:{PORT} (STRICT HTTPS / TLS)".ljust(w) + RESET)
print(CYAN + f" STORAGE : {SAVE_DIR}".ljust(w) + RESET)
print(GREEN + f" UPTIME : {uptime}s".ljust(w) + RESET)
print(GREEN + f" FILES RX : {files_received}".ljust(w) + RESET)
print(GREEN + f" DATA RX : {total_bytes} bytes".ljust(w) + RESET)
print(MAGENTA + hr("═") + RESET)
print()
except Exception:
pass
def redraw(force=False):
global last_redraw_time
now = time.monotonic()
if not force and (now - last_redraw_time < 0.2):
return
last_redraw_time = now
try:
draw_header()
for line, color in log_buffer[-400:]:
print(color + line + RESET)
sys.stdout.flush()
except Exception:
pass
def log(msg, color=WHITE, tag="INFO"):
try:
line = f"[{ts()}] [{tag}] {msg}"
log_buffer.append((line, color))
if len(log_buffer) > 1000:
log_buffer.pop(0)
redraw()
except Exception:
pass
def log_exception(e, context=""):
try:
log(f"EXCEPTION in {context}: {e}", RED, "EXC")
tb = traceback.format_exc()
for l in tb.splitlines():
log(l, RED, "TRACE")
except Exception:
pass
def cleanup_ip_tables():
now = time.monotonic()
with ban_lock:
if len(banned_ips) > 500 or len(failed_attempts) > 1000:
expired = [ip for ip, expiry in banned_ips.items() if now >= expiry]
for ip in expired:
del banned_ips[ip]
failed_attempts.pop(ip, None)
if len(failed_attempts) > 1000:
failed_attempts.clear()
def is_ip_banned(ip):
now = time.monotonic()
with ban_lock:
if ip in banned_ips:
if now < banned_ips[ip]:
return True
else:
del banned_ips[ip]
failed_attempts.pop(ip, None)
return False
def record_failed_attempt(ip):
now = time.monotonic()
cleanup_ip_tables()
with ban_lock:
count = failed_attempts.get(ip, 0) + 1
failed_attempts[ip] = count
if count >= MAX_FAILED_ATTEMPTS:
banned_ips[ip] = now + BAN_DURATION
return True
return False
def reset_ip_attempts(ip):
with ban_lock:
failed_attempts.pop(ip, None)
def set_terminal_echo(enable):
try:
import termios
if sys.stdin and sys.stdin.isatty():
fd = sys.stdin.fileno()
attrs = termios.tcgetattr(fd)
if enable:
attrs[3] |= termios.ECHO
termios.tcflush(fd, termios.TCIFLUSH)
else:
attrs[3] &= ~termios.ECHO
termios.tcsetattr(fd, termios.TCSANOW, attrs)
except Exception:
pass
def get_password_asterisks(prompt="Password: "):
try:
import msvcrt
sys.stdout.write(prompt)
sys.stdout.flush()
password = ""
while True:
ch = msvcrt.getwch()
if ch in ('\r', '\n'):
break
elif ch == '\x03':
raise KeyboardInterrupt
elif ch == '\x08':
if len(password) > 0:
password = password[:-1]
sys.stdout.write('\b \b')
sys.stdout.flush()
else:
password += ch
sys.stdout.write('*')
sys.stdout.flush()
sys.stdout.write('\n')
return password
except ImportError:
try:
import termios
import tty
sys.stdout.write(prompt)
sys.stdout.flush()
fd = sys.stdin.fileno()
old_settings = termios.tcgetattr(fd)
password = ""
try:
tty.setraw(fd)
while True:
ch = sys.stdin.read(1)
if ch in ('\r', '\n'):
break
elif ch in ('\x08', '\x7f'):
if len(password) > 0:
password = password[:-1]
sys.stdout.write('\b \b')
sys.stdout.flush()
elif ch == '\x03':
raise KeyboardInterrupt
else:
password += ch
sys.stdout.write('*')
sys.stdout.flush()
finally:
termios.tcsetattr(fd, termios.TCSADRAIN, old_settings)
sys.stdout.write('\n')
return password
except Exception:
return getpass.getpass(prompt)
def load_or_setup_password():
global saved_password_hash, saved_salt, saved_online_verifier
if os.path.exists(PASS_FILE):
try:
with open(PASS_FILE, "r") as f:
parts = f.read().strip().split(":")
if len(parts) == 3:
saved_salt = bytes.fromhex(parts[0])
saved_password_hash = bytes.fromhex(parts[1])
saved_online_verifier = bytes.fromhex(parts[2])
return
else:
print(YELLOW + " [SECURITY] Legacy password format detected. Reconfiguration required." + RESET)
except Exception as e:
print(RED + f" [ERROR] Password file ({PASS_FILE}) unreadable or corrupt. Setup required." + RESET)
if not sys.stdin or not sys.stdin.isatty():
print(RED + f" [FATAL ERROR] {PASS_FILE} requires initial configuration from an interactive terminal!" + RESET)
sys.exit(1)
print(CYAN + "\n" + hr("─") + RESET)
print(YELLOW + BOLD + " 🔐 INITIAL CREDENTIAL CONFIGURATION (HARDENED STORAGE)" + RESET)
print(WHITE + f" Maximum 512 characters for compatibility." + RESET)
while True:
pwd = get_password_asterisks(CYAN + " Create Server Password: " + RESET)
if len(pwd) == 0:
continue
if len(pwd) > 512:
print(RED + " Password too long! 512-character limit." + RESET)
continue
pwd_conf = get_password_asterisks(CYAN + " Repeat Server Password: " + RESET)
if pwd == pwd_conf:
saved_salt = os.urandom(32)
saved_password_hash = hashlib.pbkdf2_hmac('sha256', pwd.encode('utf-8'), saved_salt, 250000)
saved_online_verifier = hashlib.sha256(saved_salt + pwd.encode('utf-8')).digest()
try:
flags = os.O_WRONLY | os.O_CREAT | os.O_TRUNC
mode = 0o600
fd = os.open(PASS_FILE, flags, mode)
with os.fdopen(fd, "w") as f:
f.write(f"{saved_salt.hex()}:{saved_password_hash.hex()}:{saved_online_verifier.hex()}")
print(GREEN + " ✅ Configuration saved securely (PBKDF2/SHA-256 + 0600 Permissions)." + RESET)
time.sleep(1.0)
break
except Exception as e:
print(RED + f" Unable to save in {PASS_FILE}: {e}" + RESET)
sys.exit(1)
else:
print(RED + " Passwords do not match. Try again!\n" + RESET)
def ensure_certificates():
if not (os.path.exists(CERT_FILE) and os.path.exists(KEY_FILE)):
try:
cmd = [
"openssl", "req", "-x509", "-newkey", "ec", "-pkeyopt", "ec_paramgen_curve:prime256v1",
"-keyout", KEY_FILE, "-out", CERT_FILE,
"-days", "36500", "-nodes", "-subj", "/CN=zMPanicServer"
]
subprocess.run(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=True)
print(GREEN + " [SECURITY] New TOFU identity seeded (100-year validity). Do NOT delete .cert.pem / .key.pem!" + RESET)
return True
except Exception:
return False
return True
def has_sufficient_disk_space():
try:
st = os.statvfs(SAVE_DIR)
free_bytes = st.f_bavail * st.f_frsize
free_inodes = st.f_favail
required_reserve = MIN_FREE_DISK_BYTES + (MAX_CONCURRENT * MAX_FILE_SIZE)
if free_bytes < required_reserve or free_inodes < 1000:
return False
return True
except Exception:
return False # Fail-closed on statvfs failure
def get_safe_destination_path(raw_filename, client_ip):
uid = uuid.uuid4().hex[:8]
timestamp = time.strftime("%Y%m%d_%H%M%S")
clean_ip = client_ip.replace(":", "_").replace(".", "_")
if raw_filename:
clean_name = os.path.basename(raw_filename).replace("\x00", "").strip()
clean_name = re.sub(r'[^a-zA-Z0-9_.-]', '_', clean_name)
if clean_name and not clean_name.startswith('.'):
base_name, ext = os.path.splitext(clean_name)
final_ext = ext if ext in [".mp4", ".mkv", ".mov", ".ts", ".avi"] else ".mp4"
final_filename = f"{base_name[:64]}_{clean_ip}_{uid}{final_ext}"
else:
final_filename = f"SOS_{timestamp}_{clean_ip}_{uid}.mp4"
else:
final_filename = f"SOS_{timestamp}_{clean_ip}_{uid}.mp4"
abs_storage = os.path.abspath(SAVE_DIR)
target_path = os.path.abspath(os.path.join(abs_storage, final_filename))
if os.path.commonpath([abs_storage, target_path]) != abs_storage:
target_path = os.path.join(abs_storage, f"SOS_{timestamp}_{clean_ip}_{uid}.mp4")
return target_path
def prompt_autostart_setup():
if not sys.stdin or not sys.stdin.isatty():
return
try:
chk = subprocess.run(["crontab", "-l"], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
current_cron = chk.stdout if chk.returncode == 0 else ""
if "zmpserver" in current_cron:
return
is_frozen = getattr(sys, 'frozen', False)
if is_frozen:
exec_target = os.path.abspath(sys.executable)
work_dir = os.path.dirname(exec_target)
launch_cmd = exec_target
else:
py_script = os.path.abspath(__file__)
work_dir = os.path.dirname(py_script)
launch_cmd = f"{sys.executable} {py_script}"
screen_bin = shutil.which("screen") or "/usr/bin/screen"
print(CYAN + "\n" + hr("─") + RESET)
print(YELLOW + BOLD + " ⚙️ AUTOMATIC START CONFIGURATION (AUTOSTART AT BOOT)" + RESET)
print(WHITE + f" Detected target : {launch_cmd}" + RESET)
print(WHITE + f" Base directory : {work_dir}" + RESET)
print(CYAN + hr("─") + RESET)
print(GREEN + " Do you want to install autostart with GNU Screen [screen -S zmpserver]? (y/N) [5s timeout]: " + RESET, end="", flush=True)
rlist, _, _ = select.select([sys.stdin], [], [], 5.0)
if rlist:
ans = sys.stdin.readline().strip().lower()
else:
print(YELLOW + "\n ⏱️ Timeout (5s) expired: starting server..." + RESET)
ans = 'n'
if ans in ['s', 'si', 'y', 'yes']:
cron_entry = f"@reboot sleep 5 && cd {work_dir} && {screen_bin} -dmS zmpserver {launch_cmd}\n"
new_cron = current_cron.rstrip() + "\n" + cron_entry if current_cron else cron_entry
p = subprocess.Popen(["crontab", "-"], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
_, err = p.communicate(input=new_cron)
if p.returncode == 0:
print(GREEN + " ✅ Autostart configured successfully!" + RESET)
time.sleep(1.5)
else:
print(RED + f" ❌ Error installing to crontab: {err}" + RESET)
time.sleep(1.5)
except Exception as e:
log_exception(e, "AUTOSTART_SETUP")
def prompt_network_awake_setup():
if not sys.stdin or not sys.stdin.isatty():
return
try:
print(CYAN + "\n" + hr("─") + RESET)
print(YELLOW + BOLD + " ⚡ NETWORK AWAKE CONFIGURATION (DISABLE EEE & SLEEP)" + RESET)
print(WHITE + " EU energy rules may put the Ethernet port to sleep after days of inactivity." + RESET)
print(WHITE + " This disables EEE (ethtool), masks sleep targets, and adds a 1-min keep-alive ping." + RESET)
print(CYAN + hr("─") + RESET)
print(GREEN + " Do you want to apply network keep-awake patches? (Requires sudo) (y/N) [5s timeout]: " + RESET, end="", flush=True)
rlist, _, _ = select.select([sys.stdin], [], [], 5.0)
if rlist:
ans = sys.stdin.readline().strip().lower()
else:
print(YELLOW + "\n ⏱️ Timeout (5s) expired: skipping network patch..." + RESET)
ans = 'n'
if ans in ['s', 'si', 'y', 'yes']:
print(WHITE + " Executing patches (you may be prompted for sudo password)..." + RESET)
subprocess.run(["sudo", "ethtool", "--set-eee", "eth0", "eee", "off"], stderr=subprocess.DEVNULL)
subprocess.run(["sudo", "systemctl", "mask", "sleep.target", "suspend.target", "hibernate.target", "hybrid-sleep.target"], stderr=subprocess.DEVNULL)
chk = subprocess.run(["crontab", "-l"], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
current_cron = chk.stdout if chk.returncode == 0 else ""
cron_entry = "* * * * * ping -c 1 8.8.8.8 >/dev/null 2>&1\n"
if "ping -c 1" not in current_cron:
new_cron = current_cron.rstrip() + "\n" + cron_entry if current_cron else cron_entry
p = subprocess.Popen(["crontab", "-"], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
p.communicate(input=new_cron)
print(GREEN + " ✅ Network keep-awake patches applied successfully!" + RESET)
time.sleep(1.5)
except Exception as e:
log_exception(e, "NETWORK_AWAKE_SETUP")
def prompt_iptables_setup():
if not sys.stdin or not sys.stdin.isatty():
return
try:
print(CYAN + "\n" + hr("─") + RESET)
print(YELLOW + BOLD + " 🛡️ ENTERPRISE ANTI-DDOS & FIREWALL CONFIGURATION (IPTABLES)" + RESET)
print(WHITE + f" Applies atomic kernel-level filtering with loopback safety on port {PORT}." + RESET)
print(RED + " WARNING: Strict kernel-level packet dropping may affect unstable connections." + RESET)
print(CYAN + hr("─") + RESET)
print(GREEN + " Do you want to apply strict iptables rules? (Requires sudo) (y/N) [5s timeout]: " + RESET, end="", flush=True)
rlist, _, _ = select.select([sys.stdin], [], [], 5.0)
if rlist:
ans = sys.stdin.readline().strip().lower()
else:
print(YELLOW + "\n ⏱️ Timeout (5s) expired: skipping firewall setup..." + RESET)
ans = 'n'
if ans in ['s', 'si', 'y', 'yes']:
print(WHITE + " Configuring dedicated ZMPANIC_IN chain..." + RESET)
subprocess.run(["sudo", "iptables", "-N", "ZMPANIC_IN"], stderr=subprocess.DEVNULL)
subprocess.run(["sudo", "iptables", "-F", "ZMPANIC_IN"], stderr=subprocess.DEVNULL)
chk_jump = subprocess.run(["sudo", "iptables", "-C", "INPUT", "-p", "tcp", "--dport", str(PORT), "-j", "ZMPANIC_IN"], stderr=subprocess.DEVNULL)
if chk_jump.returncode != 0:
subprocess.run(["sudo", "iptables", "-I", "INPUT", "1", "-p", "tcp", "--dport", str(PORT), "-j", "ZMPANIC_IN"], stderr=subprocess.DEVNULL)
chain_rules = [
["sudo", "iptables", "-A", "ZMPANIC_IN", "-i", "lo", "-j", "ACCEPT"],
["sudo", "iptables", "-A", "ZMPANIC_IN", "-m", "conntrack", "--ctstate", "ESTABLISHED,RELATED", "-j", "ACCEPT"],
["sudo", "iptables", "-A", "ZMPANIC_IN", "-m", "conntrack", "--ctstate", "INVALID", "-j", "DROP"],
["sudo", "iptables", "-A", "ZMPANIC_IN", "-p", "tcp", "-m", "state", "--state", "NEW", "-m", "recent", "--set", "--name", "ZMP"],
["sudo", "iptables", "-A", "ZMPANIC_IN", "-p", "tcp", "-m", "state", "--state", "NEW", "-m", "recent", "--update", "--seconds", "60", "--hitcount", "25", "--name", "ZMP", "-j", "DROP"],
["sudo", "iptables", "-A", "ZMPANIC_IN", "-p", "tcp", "--syn", "-m", "limit", "--limit", "15/s", "--limit-burst", "30", "-j", "ACCEPT"],
["sudo", "iptables", "-A", "ZMPANIC_IN", "-p", "tcp", "--syn", "-j", "DROP"],
["sudo", "iptables", "-A", "ZMPANIC_IN", "-j", "ACCEPT"]
]
for cmd in chain_rules:
subprocess.run(cmd, stderr=subprocess.DEVNULL)
print(GREEN + " ✅ Kernel-level Anti-DDoS firewall active & loopback safe!" + RESET)
time.sleep(1.5)
except Exception as e:
log_exception(e, "FIREWALL_SETUP")
try:
if not os.path.exists(SAVE_DIR):
os.makedirs(SAVE_DIR, exist_ok=True)
except Exception as e:
log_exception(e, "INIT")
prompt_autostart_setup()
prompt_network_awake_setup()
prompt_iptables_setup()
load_or_setup_password()
splash = [
CYAN + BOLD + "███████╗███╗ ███╗██████╗ █████╗ ███╗ ██╗██╗ ██████╗" + RESET,
CYAN + BOLD + "╚══███╔╝████╗ ████║██╔══██╗██╔══██╗████╗ ██║██║██╔════╝" + RESET,
CYAN + BOLD + " ███╔╝ ██╔████╔██║██████╔╝███████║██╔██╗ ██║██║██║ " + RESET,
CYAN + BOLD + " ███╔╝ ██║╚██╔╝██║██╔═══╝ ██╔══██║██║╚██╗██║██║██║ " + RESET,
CYAN + BOLD + "███████╗██║ ╚═╝ ██║██║ ██║ ██║██║ ╚████║██║╚██████╗" + RESET,
CYAN + BOLD + "╚══════╝╚═╝ ╚═╝╚═╝ ╚═╝ ╚═╝╚═╝ ╚═══╝╚═╝ ╚═════╝" + RESET,
"",
BLUE + "Secure Evidence Ingestion Service (Hardened TLS Node)" + RESET,
WHITE + "Operational Node" + RESET,
"",
WHITE + f"Listening on 0.0.0.0:{PORT} (Strict HTTPS/TLS Only)" + RESET,
WHITE + f"Storage: {SAVE_DIR}" + RESET,
]
try:
center_block(splash)
time.sleep(1.0)
except Exception:
pass
print(CLEAR, end="")
boot_lines = []
def safe_line(expr, label):
try:
expr()
return GREEN + label + " OK" + RESET
except Exception as e:
log_exception(e, label)
return RED + label + " FAIL (see log)" + RESET
boot_lines.append(safe_line(lambda: ensure_certificates(), "Auto-SSL Certificate Provisioning..."))
boot_lines.append(safe_line(lambda: os.urandom(64), "Cryptographic subsystem.............."))
boot_lines.append(safe_line(lambda: os.path.abspath(SAVE_DIR), "Storage subsystem mount.............."))
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as test_sock:
test_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
boot_lines.append(safe_line(lambda: test_sock.bind(("0.0.0.0", PORT)), "Network stack ready.................."))
box("SYSTEM BOOT", boot_lines)
time.sleep(0.3)
class HardenedSOSHandler(http.server.BaseHTTPRequestHandler):
def log_message(self, format, *args):
pass
def setup(self):
super().setup()
self.connection.settimeout(SOCKET_TIMEOUT)
def handle(self):
try:
super().handle()
except (ConnectionResetError, BrokenPipeError, ssl.SSLError, socket.timeout, OSError):
pass
def do_POST(self):
global files_received, total_bytes, saved_salt, saved_password_hash, saved_online_verifier
part_filename = None
final_filename = None
meta_path = None
upload_start = time.monotonic()
last_chunk_time = upload_start
sock_id = id(self.connection)
try:
client_ip = self.client_address[0]
if is_ip_banned(client_ip):
self.send_response(403)
self.end_headers()
return
if self.path != "/upload":
self.send_response(404)
self.end_headers()
return
client_auth = self.headers.get("Authorization", "")
if not client_auth:
if record_failed_attempt(client_ip):
log(f"IP {client_ip} BANNED (Too many empty attempts)", RED, "BAN")
else:
log(f"ACCESS DENIED: Key Missing from IP {client_ip}", RED, "ALERT")
self.send_response(403)
self.end_headers()
return
client_online_verifier = hashlib.sha256(saved_salt + client_auth.encode('utf-8')).digest()
if not hmac.compare_digest(client_online_verifier, saved_online_verifier):
if record_failed_attempt(client_ip):
log(f"IP {client_ip} BANNED (Brute Force Detected)", RED, "BAN")
else:
log(f"ACCESS DENIED: Key Incorrect from IP {client_ip}", RED, "ALERT")
self.send_response(403)
self.end_headers()
return
reset_ip_attempts(client_ip)
if not has_sufficient_disk_space():
log(f"REJECTED PAYLOAD from {client_ip}: Disk or Inodes safety threshold hit!", RED, "STORAGE")
self.send_response(507)
self.end_headers()
return
# Client authenticated and entering active upload: switch state in watchdog
with active_sockets_lock:
if sock_id in active_sockets:
active_sockets[sock_id]["uploading"] = True
active_sockets[sock_id]["start_time"] = time.monotonic()
header_filename = self.headers.get("File-Name", "")
latitude = self.headers.get("GPS-Latitude", "0.0").strip()[:32]
longitude = self.headers.get("GPS-Longitude", "0.0").strip()[:32]
final_filename = get_safe_destination_path(header_filename, client_ip)
part_filename = final_filename + ".part"
try:
content_length = int(self.headers.get("Content-Length", 0))
except ValueError:
content_length = 0
if content_length < 0 or content_length > MAX_FILE_SIZE:
log(f"REJECTED INVALID / OVERSIZED PAYLOAD from {client_ip} ({content_length} bytes)", RED, "DROP")
self.send_response(413)
self.end_headers()
return
log(f"RECEIVING [HTTPS] from {client_ip} -> {os.path.basename(final_filename)}", BLUE, "RX")
received = 0
hasher_sha256 = hashlib.sha256()
hasher_md5 = hashlib.md5()
fd = os.open(part_filename, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
with os.fdopen(fd, "wb") as f:
while True:
now = time.monotonic()
elapsed = now - upload_start
if elapsed > MAX_UPLOAD_TIME:
raise TimeoutError("Global upload wall-clock timeout exceeded")
if elapsed > 15.0 and (received / elapsed) < MIN_BYTE_RATE:
raise TimeoutError("Upload throughput dropped below minimum threshold")
if (now - last_chunk_time) > SOCKET_TIMEOUT:
raise TimeoutError("Inter-read idle timeout exceeded")
remaining = (content_length - received) if content_length > 0 else (MAX_FILE_SIZE - received)
if remaining <= 0:
break
chunk_limit = min(16384, remaining)
chunk = self.rfile.read1(chunk_limit)
if not chunk:
break
f.write(chunk)
hasher_sha256.update(chunk)
hasher_md5.update(chunk)
received += len(chunk)
last_chunk_time = time.monotonic()
f.flush()
os.fsync(f.fileno())
if received <= 0 or (content_length > 0 and received != content_length):
log(f"INCOMPLETE STREAM from {client_ip} ({received}/{content_length} bytes)", RED, "WARN")
if os.path.exists(part_filename):
try:
os.remove(part_filename)
except Exception:
pass
self.send_response(400)
self.end_headers()
return
file_sha256 = hasher_sha256.hexdigest()
file_md5 = hasher_md5.hexdigest()
meta_path = os.path.splitext(final_filename)[0] + ".meta.txt"
meta_fd = os.open(meta_path + ".part", os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
with os.fdopen(meta_fd, "w", encoding="utf-8") as meta_f:
meta_f.write(f"File: {os.path.basename(final_filename)}\n")
meta_f.write(f"Protocol: HTTPS (TLS Encrypted)\n")
meta_f.write(f"Client-IP: {client_ip}\n")
meta_f.write(f"Timestamp: {ts()}\n")
meta_f.write(f"GPS-Latitude: {latitude}\n")
meta_f.write(f"GPS-Longitude: {longitude}\n")
meta_f.write(f"Maps-Link: https://maps.google.com/maps?q={latitude},{longitude}\n")
meta_f.write(f"SHA-256: {file_sha256}\n")
meta_f.write(f"MD5: {file_md5}\n")
meta_f.flush()
os.fsync(meta_f.fileno())
# Atomic publication: evidence file is never visible incomplete
os.replace(part_filename, final_filename)
os.replace(meta_path + ".part", meta_path)
files_received += 1
total_bytes += received
self.send_response(200)
self.send_header("Content-Type", "text/plain")
self.end_headers()
self.wfile.write(b"SUCCESS: Secured.")
log(f"STORED [HTTPS]: {os.path.basename(final_filename)} ({received} bytes)", GREEN, "OK")
except Exception as e:
log_exception(e, "do_POST")
# Always clean up partial evidence on ANY failure
if part_filename and os.path.exists(part_filename):
try:
os.remove(part_filename)
except Exception:
pass
if meta_path and os.path.exists(meta_path + ".part"):
try:
os.remove(meta_path + ".part")
except Exception:
pass
try:
self.send_response(500)
self.end_headers()
except Exception:
pass
def signal_handler(sig, frame):
global shutdown_flag
set_terminal_echo(True)
log("SHUTDOWN SIGNAL RECEIVED. STOPPING CLEANLY...", YELLOW, "SYS")
shutdown_flag = True
signal.signal(signal.SIGINT, signal_handler)
signal.signal(signal.SIGTERM, signal_handler)
ensure_certificates()
log("NODE ONLINE (STRICT HTTPS / TLS). READY FOR CONNECTIONS...", CYAN, "SYS")
log("TOFU LOCK ACTIVE: Do not delete .cert.pem/.key.pem or paired clients will reject connection!", YELLOW, "SEC")
redraw(force=True)
set_terminal_echo(False)
class SecureThreadingServer(ThreadingMixIn, HTTPServer):
daemon_threads = True
block_on_close = False
def __init__(self, server_address, RequestHandlerClass, ssl_context):
super().__init__(server_address, RequestHandlerClass)
self.socket = ssl_context.wrap_socket(self.socket, server_side=True, do_handshake_on_connect=False)
def process_request(self, request, client_address):
# Strict admission control BEFORE thread creation: Prevents unbounded thread spawning DoS
acquired = concurrency_semaphore.acquire(blocking=False)
if not acquired:
try:
request.shutdown(socket.SHUT_RDWR)
except Exception:
pass
try:
request.close()
except Exception:
pass
return # No thread spawned
# Concurrency slot reserved: spawn bounded worker thread
t = threading.Thread(target=self.process_request_thread_guarded, args=(request, client_address))
t.daemon = True
t.start()
def process_request_thread_guarded(self, request, client_address):
sock_id = id(request)
with active_sockets_lock:
active_sockets[sock_id] = {
"sock": request,
"start_time": time.monotonic(),
"uploading": False
}
try:
request.settimeout(SOCKET_TIMEOUT)
request.do_handshake()
self.finish_request(request, client_address)
except Exception:
pass
finally:
with active_sockets_lock:
active_sockets.pop(sock_id, None)
try:
request.shutdown(socket.SHUT_RDWR)
except Exception:
pass
try:
request.close()
except Exception:
pass
# Worker and ONLY worker releases its own acquired token
concurrency_semaphore.release()
def handle_error(self, request, client_address):
pass
def nuclear_watchdog():
while not shutdown_flag:
time.sleep(5.0)
now = time.monotonic()
with active_sockets_lock:
for sock_id, meta in list(active_sockets.items()):
sock = meta["sock"]
start = meta["start_time"]
is_uploading = meta["uploading"]
max_allowed = MAX_UPLOAD_TIME if is_uploading else HEADER_TIMEOUT
if now - start > max_allowed:
# Watchdog forces socket closure. Does NOT touch semaphore.
try:
sock.shutdown(socket.SHUT_RDWR)
except Exception:
pass
try:
sock.close()
except Exception:
pass
def eee_keepalive_pulse():
targets = [
("1.1.1.1", 53),
("8.8.8.8", 53),
("9.9.9.9", 53),
("208.67.222.222", 53),
("192.168.1.1", 53),
("192.168.0.1", 53),
("10.0.0.1", 53)
]
while not shutdown_flag:
for ip, port in targets:
try:
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s:
s.settimeout(1.0)
s.sendto(b"\x00", (ip, port))
break
except Exception:
continue
time.sleep(15.0)
eee_thread = threading.Thread(target=eee_keepalive_pulse, daemon=True)
eee_thread.start()
try:
import resource
soft, hard = resource.getrlimit(resource.RLIMIT_NOFILE)
resource.setrlimit(resource.RLIMIT_NOFILE, (min(hard, 4096), hard))
except Exception:
pass
watchdog_thread = threading.Thread(target=nuclear_watchdog, daemon=True)
watchdog_thread.start()
while not shutdown_flag:
httpd = None
try:
ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
ssl_context.minimum_version = ssl.TLSVersion.TLSv1_2
ssl_context.load_cert_chain(certfile=CERT_FILE, keyfile=KEY_FILE)
httpd = SecureThreadingServer(("0.0.0.0", PORT), HardenedSOSHandler, ssl_context)
server_thread = threading.Thread(target=httpd.serve_forever, daemon=True)
server_thread.start()
while not shutdown_flag:
if not server_thread.is_alive():
log("SUPERVISOR: Main engine thread died silently. Triggering restart...", RED, "LIFEBOAT")
break
time.sleep(1.0)
except KeyboardInterrupt:
shutdown_flag = True
break
except Exception as e:
if not shutdown_flag:
log_exception(e, "SUPERVISOR_WATCHDOG")
time.sleep(1.0)
finally:
set_terminal_echo(True)
if httpd:
try:
httpd.shutdown()
httpd.server_close()
except Exception:
pass
log("SERVER TERMINATED SAFELY.", YELLOW, "SYS")| Feature | Implementation |
|---|---|
| Dependencies | Zero (pip not required) |
| Transport | Strict HTTPS / TLS only (auto-provisioned RSA-2048 self-signed cert) |
| Authentication | Zero-knowledge PBKDF2-HMAC-SHA256 (250 000 iterations) + 32-byte salt in .srvpass.txt |
| Comparison | Constant-time hmac.compare_digest |
| Rate limiting | In-memory; 5 consecutive failures → 1-hour IP ban |
| Architecture | Blind drop-box — only POST /upload is accepted |
| Path safety | Regex sanitization + os.path.commonpath anti-traversal |
| Naming | SOS_{timestamp}_{clean_ip}_{uid}.mp4 + sibling .meta.txt with Google Maps link |
| Resilience | Supervisor watchdog auto-restarts the TLS listener on crash |
| First-run UX | Interactive masked-asterisk password setup (max 512 chars) + optional @reboot screen autostart |
python3 SERVER.py
# First run → create password (≤ 512 characters)
# Optional: answer “s” / “y” to install GNU Screen autostart via crontabRecordings and forensic metadata land in ./zmpanic_recordings/.
| Element | Value / Description |
|---|---|
| Method / Path | POST /upload |
| Content-Type | application/octet-stream (raw binary stream) |
| Authorization | Bearer <server_password> (required) |
| File-Name | Client-generated name, e.g. SOS_1718115648000.mp4 |
| GPS-Latitude | String representation of last known latitude |
| GPS-Longitude | String representation of last known longitude |
| Content-Length | Exact byte length of the video payload |
| Success | HTTP 200 + body SUCCESS: Secured. |
| Auth failure | HTTP 401 (after 5 failures the source IP is banned for 1 hour) |
| Banned IP | HTTP 403 |
| Oversized payload | HTTP 413 (server limit 100 MB) |
| Any other method/path | HTTP 404 or 405 |
The Android client constructs the target list from a semicolon-delimited field and always prefers HTTPS. The first successful 2xx response terminates further attempts for that chunk.
zMPanic/
├── app/
│ └── src/main/
│ ├── java/com/mykerd/panic/
│ │ ├── MainActivity.kt # Control surface, permissions, password dialog
│ │ └── PanicService.kt # Dual-mode recorder, Ghost Engine, store-and-forward
│ ├── res/
│ │ ├── values/ # Default (English) strings & resources
│ │ ├── values-*/ # 62+ native locale overrides
│ │ ├── layout/
│ │ ├── drawable/
│ │ └── xml/
│ └── AndroidManifest.xml
├── gradle/
├── SERVER.py # Hardened TLS ingestion node (zero dependencies)
├── build.gradle.kts
├── settings.gradle.kts
├── gradle.properties
├── gradlew / gradlew.bat
├── LICENSE
├── PRIVACY_POLICY.md
├── ic_launcher.svg
└── README.md # This document
The zM SOS GUARD protocol is structurally and philosophically incompatible with iOS for the following technical reasons:
-
Background Camera Kill
iOS immediately suspendsAVCaptureSessionthe moment the app loses foreground focus, is minimized, or the screen turns off. There is no equivalent to Android’s high-priority Foreground Service that permits continuous camera + microphone access while the device is locked. -
Hardware Privacy Indicators
The orange/green status-bar privacy dots are wired directly to the camera and microphone hardware. They cannot be suppressed or disguised by software, rendering any stealth mode useless. -
Sideloading & Distribution Autonomy
Android permits native compilation and immediate installation of a standalone APK. Apple requires a Mac, a paid developer account, and enforces a 7-day expiration on unsigned or free-provisioning builds, destroying the “deploy anywhere, anytime” operational model required for emergency black-box use.
Consequently, the entire architecture — continuous background recording, process-level network binding, stealth notification channels, and long-running foreground services — exists only on Android.
This software is a tool for forensic protection, emergency documentation, and technical research into the resilience of mobile operating systems. It is released exclusively for legitimate security and personal-protection purposes.
The user assumes full and total civil and criminal responsibility arising from the use of hidden recording features. The user must comply strictly with all applicable local laws governing privacy, personal data processing, and the interception of communications, images, and audio.
The developer and project contributors assume no responsibility for damages, improper use, illicit actions, or unauthorized applications of the source code documented herein.
Evidence produced by this system should be treated according to local chain-of-custody and forensic best practices. The non-destructive retention model (.synced.mp4 flags) is intentionally designed to preserve the original files for later independent verification.
zM SOS GUARD v13.0 — The Hardened & Secure Update
Hardware-backed secrets · Zero-knowledge PBKDF2 server · Dual-mode self-healing video · Rugged OEM fixes · Store-and-forward multi-server resilience · 62+ native locales.
Pure Vanilla. Zero Dependencies. Maximum Survivability.