How to Set Up a Dead Man’s Switch for Data

#!/usr/bin/env python3
"""
Dead Man's Switch — checks if you've checked in recently.
Run via cron every hour on a server or trusted machine.
"""
import os
import json
import time
import hashlib
import smtplib
import logging
from datetime import datetime, timedelta
from email.message import EmailMessage
from pathlib import Path

# Configuration
CONFIG = {
    "check_in_file": "/var/dms/last_checkin.json",
    "state_file": "/var/dms/state.json",
    "payload_file": "/var/dms/payload.enc",       # Encrypted payload to release
    "log_file": "/var/log/dms.log",

    # Time windows (hours)
    "reminder_after_hours": 48,     # Send you a reminder after 2 days no check-in
    "alert_after_hours": 96,        # Alert trusted contacts after 4 days
    "release_after_hours": 168,     # Release payload after 7 days

    # Your email (where reminder goes)
    "owner_email": "you@protonmail.com",

    # Trusted contacts (receive alert and payload)
    "trusted_contacts": [
        {"name": "Alice", "email": "alice@example.com"},
        {"name": "Bob", "email": "bob@example.com"},
    ],

    # SMTP settings (use an email relay or local MTA)
    "smtp_host": "localhost",
    "smtp_port": 25,
    "smtp_from": "dms@yourserver.com",
    "smtp_user": "",    # leave empty for unauthenticated local relay
    "smtp_pass": "",
}

logging.basicConfig(
    filename=CONFIG["log_file"],
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(message)s"
)

def load_last_checkin():
    """Return datetime of last check-in, or epoch if no record."""
    p = Path(CONFIG["check_in_file"])
    if not p.exists():
        return datetime.fromtimestamp(0)
    data = json.loads(p.read_text())
    return datetime.fromtimestamp(data.get("timestamp", 0))

def load_state():
    p = Path(CONFIG["state_file"])
    if not p.exists():
        return {}
    return json.loads(p.read_text())

def save_state(state: dict):
    Path(CONFIG["state_file"]).write_text(json.dumps(state, indent=2))

def send_email(to: str, subject: str, body: str, attachment_path: str = None):
    msg = EmailMessage()
    msg["From"] = CONFIG["smtp_from"]
    msg["To"] = to
    msg["Subject"] = subject
    msg.set_content(body)

    if attachment_path and Path(attachment_path).exists():
        with open(attachment_path, "rb") as f:
            msg.add_attachment(f.read(), maintype="application", subtype="octet-stream",
                             filename=Path(attachment_path).name)

    try:
        with smtplib.SMTP(CONFIG["smtp_host"], CONFIG["smtp_port"]) as s:
            if CONFIG["smtp_user"]:
                s.login(CONFIG["smtp_user"], CONFIG["smtp_pass"])
            s.send_message(msg)
        logging.info(f"Email sent to {to}: {subject}")
    except Exception as e:
        logging.error(f"Failed to send email to {to}: {e}")

def run_check():
    last_checkin = load_last_checkin()
    now = datetime.now()
    hours_since = (now - last_checkin).total_seconds() / 3600
    state = load_state()

    logging.info(f"Check-in status: last={last_checkin}, hours_since={hours_since:.1f}")

    # Stage 1: Send owner reminder
    if hours_since >= CONFIG["reminder_after_hours"] and not state.get("reminder_sent"):
        send_email(
            CONFIG["owner_email"],
            "DMS: Check-in required",
            f"Your dead man's switch hasn't received a check-in in {hours_since:.0f} hours.\n"
            f"Check in now: curl -X POST https://yourserver.com/dms/checkin?token=YOUR_TOKEN\n"
            f"Next escalation in {CONFIG['alert_after_hours'] - hours_since:.0f} hours."
        )
        state["reminder_sent"] = True
        state["reminder_time"] = now.isoformat()
        save_state(state)

    # Stage 2: Alert trusted contacts
    if hours_since >= CONFIG["alert_after_hours"] and not state.get("alert_sent"):
        for contact in CONFIG["trusted_contacts"]:
            send_email(
                contact["email"],
                f"DMS Alert: {CONFIG['owner_email']} has not checked in",
                f"This is an automated alert from {CONFIG['owner_email']}'s dead man's switch.\n\n"
                f"There has been no check-in for {hours_since:.0f} hours.\n"
                f"Please attempt to contact them directly.\n\n"
                f"If no check-in occurs, encrypted data will be released in "
                f"{CONFIG['release_after_hours'] - hours_since:.0f} more hours."
            )
        state["alert_sent"] = True
        state["alert_time"] = now.isoformat()
        save_state(state)

    # Stage 3: Release encrypted payload
    if hours_since >= CONFIG["release_after_hours"] and not state.get("payload_released"):
        for contact in CONFIG["trusted_contacts"]:
            send_email(
                contact["email"],
                f"DMS: Encrypted payload from {CONFIG['owner_email']}",
                f"No check-in received for {hours_since:.0f} hours. Releasing encrypted payload.\n\n"
                f"The attached file is encrypted. Decryption instructions were given to you separately.\n"
                f"Use: gpg --decrypt payload.enc",
                attachment_path=CONFIG["payload_file"]
            )
        state["payload_released"] = True
        state["release_time"] = now.isoformat()
        save_state(state)
        logging.warning(f"PAYLOAD RELEASED after {hours_since:.0f}h without check-in")

if __name__ == "__main__":
    run_check()

#!/usr/bin/env python3
"""Simple Flask endpoint to accept check-in tokens."""
from flask import Flask, request, jsonify
import json, hashlib, time
from pathlib import Path

app = Flask(__name__)

# Set your check-in token (share only with yourself)
# Generate: python3 -c "import secrets; print(secrets.token_urlsafe(32))"
VALID_TOKEN = "YOUR_SECRET_CHECK_IN_TOKEN"
CHECK_IN_FILE = "/var/dms/last_checkin.json"

@app.route("/dms/checkin", methods=["POST", "GET"])
def checkin():
    token = request.args.get("token") or request.form.get("token")

    if not token:
        return jsonify({"error": "Missing token"}), 400

    # Constant-time comparison to prevent timing attacks
    if not hmac.compare_digest(token, VALID_TOKEN):
        return jsonify({"error": "Invalid token"}), 403

    # Record check-in
    data = {"timestamp": time.time(), "ip": request.remote_addr}
    Path(CHECK_IN_FILE).write_text(json.dumps(data))

    return jsonify({"status": "ok", "message": "Check-in recorded"}), 200

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000)

from flask import Flask, request, jsonify, abort
from functools import wraps
import time
import hmac

app = Flask(__name__)

# Simple in-memory rate limiter (use Redis for multi-process deployments)
request_times = {}

def rate_limit(max_requests=5, window_seconds=60):
    def decorator(f):
        @wraps(f)
        def wrapper(*args, **kwargs):
            ip = request.remote_addr
            now = time.time()
            times = request_times.get(ip, [])
            # Remove old timestamps outside the window
            times = [t for t in times if now - t < window_seconds]
            if len(times) >= max_requests:
                abort(429)
            times.append(now)
            request_times[ip] = times
            return f(*args, **kwargs)
        return wrapper
    return decorator

@app.route("/dms/checkin", methods=["POST", "GET"])
@rate_limit(max_requests=10, window_seconds=60)
def checkin():
    token = request.args.get("token") or request.form.get("token")
    if not token:
        return jsonify({"error": "Missing token"}), 400
    if not hmac.compare_digest(token.encode(), VALID_TOKEN.encode()):
        time.sleep(1)  # Slow down invalid attempts
        return jsonify({"error": "Invalid token"}), 403
    # ... rest of check-in logic

# Distributed check-in: sends the check-in token to multiple servers simultaneously
import requests
import concurrent.futures

DMS_ENDPOINTS = [
    "https://dms1.yourserver.com/dms/checkin",
    "https://dms2.backupserver.net/dms/checkin",
]
TOKEN = "YOUR_SECRET_TOKEN"

def checkin_server(url):
    try:
        r = requests.post(url, params={"token": TOKEN}, timeout=10)
        return url, r.status_code == 200
    except Exception as e:
        return url, False

def checkin_all():
    with concurrent.futures.ThreadPoolExecutor() as executor:
        results = list(executor.map(checkin_server, DMS_ENDPOINTS))

    all_ok = all(ok for _, ok in results)
    for url, ok in results:
        status = "OK" if ok else "FAILED"
        print(f"{status}: {url}")

    if not all_ok:
        print("WARNING: Not all DMS servers received check-in")

    return all_ok

if __name__ == "__main__":
    checkin_all()