#!/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()