sij/hand_of_morpheus
1
1
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Big revamp, new modular design, now able to delete users from conduwuit without matching registration requests, handle warrant canary, lots more

Browse source
This commit is contained in:
Sangye Ince-Johannsen 2025-04-04 08:25:18 +00:00
parent 4ce1ff8cfd
commit d536648058
32 changed files with 1176 additions and 358 deletions

20
.gitignore vendored
View file

@ -14,7 +14,7 @@ ENV/
# Log files
*.log
logs/
sw1tch/logs/
# IDE or OS-specific files
.DS_Store
@ -29,18 +29,8 @@ Thumbs.db
.classpath
# Project-specific sensitive files
.registration_token
config.yaml
registrations.json
banned_ips.txt
refresh_token.sh
testbench/
registration.pid
# Backup directories
backup/
*_backup/
*.bak
sw1tch/data/*
sw1tch/config/*
# Test cache
.pytest_cache/
@ -48,10 +38,6 @@ backup/
htmlcov/
.tox/
# Dependency management
pip-log.txt
pip-delete-this-directory.txt
# Environment variables
.env*
!.env.example

12
banned_emails.txt
View file

@ -1,12 +0,0 @@
*@yopmail.com
*@letterguard.net
*@sharklasers.com
*@msssg.com
*@10mail.org
*@monopolio.net
*@owlny.com
*@aleeas.com
*@passinbox.com
*@polkaroad.net
*@onionmail.org
*@mail2tor.com

5
banned_usernames.txt
View file

@ -1,5 +0,0 @@
.*admin.*
.*loli.*
.*shota.*
.*pedo.*
.*pthc.*

30
canary.txt
View file

@ -1,30 +0,0 @@
-----BEGIN PGP SIGNED MESSAGE-----
Hash: SHA512
We2.ee Warrant Canary · 2025-03-30T11:43:29.0985236 UTC
I, 〄, the server administrator of We2.ee, state this 30th day of March, 2025:
1. We2.ee has not received any National Security Letters or FISA court orders.
2. We2.ee has not been subject to any searches or seizures of our servers.
3. We2.ee has not been required to modify our systems to facilitate surveillance.
4. We2.ee has not received any gag order that restrain our ability to make these declarations truthfully.
Datestamp Proof:
Daily News: ""The Encampments": New Film on Mahmoud Khalil & Columbia Students Who Sparked Gaza Campus Protests"
Source URL: http://www.democracynow.org/2025/3/28/the_encampments_columbia_university_documentary
BTC block: #890107, 2025-03-30 11:39:28 UTC
Block hash: 362a1d72055578b915f653a99da326bee7c664d8b158
-----BEGIN PGP SIGNATURE-----
iQIzBAEBCgAdFiEEMjqKLEezdiJLNhO3U1smWu2+W0QFAmfpLpUACgkQU1smWu2+
W0StRw/8Cp6zZVvoVfyyWEmGyTbXbbYCmpHcHyyB+17ZZQPu3ZVHCrS58e0JTcb/
x+NF/0QfBpwsmNJk5qWqwNkdoZLc26tW47XcyzLAs1f9OoKsFf/OIFkggL2AUtoS
t+vZ/e2pW/5YTud9L4K2CrKReuEb/yzXE5PVqqN/OYwYXUzb0LvNub7o+DtVWckD
tJueplKr9jmDgxRXZ/9RBDvsJtsnQynYkFnPxHodBNCaMClshIr/XrPKT7uB97k2
n/Q3JNAMlwGSO2cKRGIhBlTshnNOeNZo+7cOA/zKbUhpZxOIb3OeQvt4exhcM4WE
ezubOMJMfd9h4IgtPY7/ZxbP+k1OFzHV6hP39xFbmgpsdPK9mnWRlvPR3Hhfzo/B
DNh0fn4JS8GLCO9SujGqJ81Pj6l3g0HzMbsdu8zDxfz2CiV4FGpOWJ1r/Gw8rKtX
J5ptp0iUbpmznDDZwHLlq9nLdYOcnbZX5PHKksbZ86H//1AMvYN9GWaT56PwCJbK
eH5o23u8HnZom0mmDS0OPOAmImNoQKkhVnnECqyrg6lPsgXGc6B7+e+fkeoQz6vy
taiq7yaumyKT8mQ3YmRiGQ4pI9wTFFQsHt5p4jZ02RAWZGMTolurlLVHhVFWEsyc
03Xm5Zzb1yfh/kWcHxu5r4uVdhCDx2W1gjE9eUKQW3EJPDS+Kn0=
=fj1f
-----END PGP SIGNATURE-----

133
cleanup.py
View file

@ -1,133 +0,0 @@
import os
import json
import yaml
import httpx
import logging
from typing import List, Dict
from datetime import datetime, timedelta
# Setup logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
# Load paths and config
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
CONFIG_PATH = os.path.join(BASE_DIR, "config.yaml")
REGISTRATIONS_PATH = os.path.join(BASE_DIR, "registrations.json")
def load_config() -> dict:
"""Load configuration from yaml file."""
with open(CONFIG_PATH, "r") as f:
return yaml.safe_load(f)
def load_registrations() -> List[Dict]:
"""Load current registrations from JSON file."""
try:
with open(REGISTRATIONS_PATH, "r") as f:
return json.load(f)
except (FileNotFoundError, json.JSONDecodeError):
return []
def save_registrations(registrations: List[Dict]):
"""Save updated registrations back to JSON file."""
with open(REGISTRATIONS_PATH, "w") as f:
json.dump(registrations, f, indent=2)
async def check_username_exists(username: str, homeserver: str) -> bool:
"""
Check if a username exists on the Matrix server.
Returns True if the username exists, False otherwise.
"""
url = f"https://{homeserver}/_matrix/client/v3/register/available?username={username}"
async with httpx.AsyncClient() as client:
try:
response = await client.get(url, timeout=5)
if response.status_code == 200:
# 200 OK with available=true means username exists
return response.json().get("available", False)
elif response.status_code == 400:
# 400 Bad Request means username is not available
return False
except httpx.RequestError as ex:
logger.error(f"Error checking username {username}: {ex}")
return False
return False
async def cleanup_registrations(min_age_hours: int = 24):
"""
Clean up registrations by removing entries for usernames that don't exist on the server,
but only if they're older than min_age_hours.
Never removes entries for existing Matrix users regardless of age.
"""
config = load_config()
registrations = load_registrations()
if not registrations:
logger.info("No registrations found to clean up")
return
logger.info(f"Starting cleanup of {len(registrations)} registrations")
logger.info(f"Will only remove non-existent users registered more than {min_age_hours} hours ago")
# Track which entries to keep
entries_to_keep = []
removed_count = 0
too_new_count = 0
exists_count = 0
current_time = datetime.utcnow()
for entry in registrations:
username = entry["requested_name"]
reg_date = datetime.fromisoformat(entry["datetime"])
age = current_time - reg_date
# First check if the user exists on Matrix
exists = await check_username_exists(username, config["homeserver"])
if exists:
# Always keep entries for existing Matrix users
entries_to_keep.append(entry)
exists_count += 1
logger.info(f"Keeping registration for existing user: {username}")
continue
# For non-existent users, check if they're old enough to remove
if age < timedelta(hours=min_age_hours):
# Keep young entries even if user doesn't exist yet
entries_to_keep.append(entry)
too_new_count += 1
logger.info(f"Keeping recent registration: {username} (age: {age.total_seconds()/3600:.1f} hours)")
else:
# Remove old entries where user doesn't exist
logger.info(f"Removing old registration: {username} (age: {age.total_seconds()/3600:.1f} hours)")
removed_count += 1
# Save updated registrations
save_registrations(entries_to_keep)
logger.info(f"Cleanup complete:")
logger.info(f"- Kept {exists_count} entries for existing Matrix users")
logger.info(f"- Kept {too_new_count} entries younger than {min_age_hours} hours")
logger.info(f"- Removed {removed_count} old entries for non-existent users")
logger.info(f"- Total remaining entries: {len(entries_to_keep)}")
if __name__ == "__main__":
import asyncio
import argparse
parser = argparse.ArgumentParser(description="Clean up Matrix registration entries")
parser.add_argument(
"--min-age-hours",
type=int,
default=24,
help="Minimum age in hours before removing non-existent users (default: 24)"
)
args = parser.parse_args()
asyncio.run(cleanup_registrations(args.min_age_hours))

36
conduwuit.env
View file

@ -1,36 +0,0 @@
# conduwuit.env
CONDUWUIT_SERVER_NAME=we2.ee
CONDUWUIT_DATABASE_PATH=/var/lib/conduwuit/conduwuit.db
CONDUWUIT_DATABASE_BACKEND=rocksdb
CONDUWUIT_DATABASE_BACKUP_PATH=/backup
CONDUWUIT_ALLOW_REGISTRATION=true
CONDUWUIT_REGISTRATION_TOKEN_FILE=/.registration_token
CONDUWUIT_ADDRESS=["0.0.0.0", "::"]
CONDUWUIT_PORT=8008
CONDUWUIT_NEW_USER_DISPLAYNAME_SUFFIX=
CONDUWUIT_AUTO_JOIN_ROOMS=["#home:we2.ee", "#server:we2.ee", "#pub:we2.ee", "#help:we2.ee"]
CONDUWUIT_DB_CACHE_CAPACITY_MB=1024
CONDUWUIT_DB_WRITE_BUFFER_CAPACITY_MB=256
CONDUWUIT_DB_POOL_WORKERS=64
CONDUWUIT_DB_POOL_WORKERS_LIMIT=128
CONDUWUIT_STREAM_AMPLIFICATION=8192
CONDUWUIT_MAX_REQUEST_SIZE=33554432
CONDUWUIT_CACHE_CAPACITY_MODIFIER=1.5
CONDUWUIT_ALLOW_FEDERATION=true
CONDUWUIT_ALLOW_PUBLIC_ROOM_DIRECTORY_OVER_FEDERATION=true
CONDUWUIT_ALLOW_PUBLIC_ROOM_DIRECTORY_WITHOUT_AUTH=true
CONDUWUIT_WELL_KNOWN_CONN_TIMEOUT=30
CONDUWUIT_FEDERATION_TIMEOUT=60
CONDUWUIT_FEDERATION_IDLE_TIMEOUT=60
CONDUWUIT_SENDER_TIMEOUT=600
CONDUWUIT_SENDER_IDLE_TIMEOUT=300
CONDUWUIT_SENDER_SHUTDOWN_TIMEOUT=30
CONDUWUIT_DNS_CACHE_ENTRIES=0
CONDUWUIT_DNS_MIN_TTL=0
CONDUWUIT_DNS_MIN_TTL_NXDOMAIN=15
CONDUWUIT_DNS_ATTEMPTS=5
CONDUWUIT_DNS_TIMEOUT=5
CONDUWUIT_DNS_TCP_FALLBACK=true
CONDUWUIT_QUERY_ALL_NAMESERVERS=false
CONDUWUIT_QUERY_OVER_TCP_ONLY=false
CONDUWUIT_IP_LOOKUP_STRATEGY=3

47
example-config.yaml
View file

@ -1,47 +0,0 @@
port: 6626
homeserver: "we2.ee"
# Token reset configuration
token_reset_time_utc: 0 # 00:00 UTC
downtime_before_token_reset: 10 # 10 minutes before that time, registration is closed
# Email rate limiting and multiple account settings
email_cooldown: 3600 # 1 hour cooldown between requests for the same email
multiple_users_per_email: false # If false, each email can only be used once
# Matrix admin credentials
matrix:
enabled: true
homeserver: "https://we2.ee"
room_id: "!Announcements_Room_ID:we2.ee"
username: "@canary:we2.ee"
password: "Password_of_canary"
# GPG Configuration
gpg:
key_id: "Your_GPG_Key_ID"
# SMTP configuration
smtp:
host: "smtp.protonmail.ch"
port: 587
username: "admin@we2.ee"
password: "YourPassword"
use_tls: true
# Email templates
email_subject: "Your Registration Token for {homeserver}"
email_body: |
Hello,
Thank you for your interest in {homeserver}, {requested_username}.
The registration token today is: {registration_token}
This registration token is valid for {time_until_reset}. If you do not register in that period, you will need to request the new registration token.
Please ensure you use the username {requested_username} when you register. Using a different username may result in your account being deleted at a later time without forewarning.
Regards,
{homeserver} registration team

36
launch_conduwuit.sh
View file

@ -3,11 +3,11 @@
# File paths
BASE_PATH="/home/sij/hand_of_morpheus"
TOKEN_FILE="$BASE_PATH/.registration_token"
LOG_FILE="$BASE_PATH/token_refresh.log"
LOG_FILE="$BASE_PATH/logs/token_refresh.log"
BACKUP_PATH="/home/sij/conduwuit_backup"
ENV_FILE="$BASE_PATH/conduwuit.env"
ENV_FILE="$BASE_PATH/config/conduwuit.env"
REPO_PATH="$HOME/workshop/conduwuit"
CONFIG_FILE="$BASE_PATH/config.yaml"
CONFIG_FILE="$BASE_PATH/config/config.yaml"
# Static container settings
CONTAINER_NAME="conduwuit"
@ -41,32 +41,27 @@ refresh_token() {
update_docker_image() {
log "Updating Conduwuit Docker image..."
# Navigate to the repository directory
cd "$REPO_PATH" || {
log "ERROR: Failed to cd into $REPO_PATH"
exit 1
}
# Pull the latest changes
git pull origin main || {
log "ERROR: git pull failed"
exit 1
}
# Build the Docker image using Nix
nix build -L --extra-experimental-features "nix-command flakes" .#oci-image-x86_64-linux-musl-all-features || {
log "ERROR: nix build failed"
exit 1
}
# Use the result symlink to find the image tarball
IMAGE_TAR_PATH=$(readlink -f result)
if [ ! -f "$IMAGE_TAR_PATH" ]; then
log "ERROR: No image tarball found at $IMAGE_TAR_PATH"
exit 1
fi
# Load the image into Docker and tag it
docker load < "$IMAGE_TAR_PATH" | awk '/Loaded image:/ { print $3 }' | xargs -I {} docker tag {} "$CONTAINER_IMAGE"
if [ $? -ne 0 ]; then
log "ERROR: Failed to load and tag Docker image"
@ -77,11 +72,9 @@ update_docker_image() {
# Function to restart the container
restart_container() {
# Stop and remove existing container
docker stop "$CONTAINER_NAME" 2>/dev/null
docker rm "$CONTAINER_NAME" 2>/dev/null
# Base docker run command
DOCKER_CMD=(docker run -d
-v "db:/var/lib/conduwuit/"
-v "${TOKEN_FILE}:/.registration_token:ro"
@ -91,12 +84,9 @@ restart_container() {
--restart unless-stopped
)
# Read the .env file and append CONDUWUIT_ variables as -e options
if [ -f "$ENV_FILE" ]; then
while IFS='=' read -r key value; do
# Skip empty lines and comments
[[ -z "$key" || "$key" =~ ^# ]] && continue
# Trim whitespace
key=$(echo "$key" | xargs)
value=$(echo "$value" | xargs)
if [[ "$key" =~ ^CONDUWUIT_ ]]; then
@ -109,23 +99,17 @@ restart_container() {
exit 1
fi
# Add RUST_LOG explicitly (since it’s not CONDUWUIT_ prefixed)
DOCKER_CMD+=(-e RUST_LOG="conduwuit=trace,reqwest=trace,hickory_proto=trace")
# Add emergency password if --super-admin is set
if [ "$SUPER_ADMIN" = true ]; then
EMERGENCY_PASSWORD=$(openssl rand -hex 8)
log "Setting emergency password to: $EMERGENCY_PASSWORD"
DOCKER_CMD+=(-e CONDUWUIT_EMERGENCY_PASSWORD="$EMERGENCY_PASSWORD")
fi
# Add the image as the last argument
DOCKER_CMD+=("$CONTAINER_IMAGE")
# Log the full command for debugging
log "Docker command: ${DOCKER_CMD[*]}"
# Execute the docker command
"${DOCKER_CMD[@]}"
if [ $? -ne 0 ]; then
log "ERROR: Failed to create new conduwuit container"
@ -135,7 +119,6 @@ restart_container() {
log "Successfully recreated container \"$CONTAINER_NAME\" with image \"$CONTAINER_IMAGE\"."
log " - Configuration loaded from $ENV_FILE"
# Log super-admin credentials if applicable
if [ "$SUPER_ADMIN" = true ]; then
log "Use the following credentials to log in as the @conduit server user:"
log " Username: @conduit:we2.ee"
@ -144,19 +127,17 @@ restart_container() {
fi
}
# Function to ensure the registration service is running.
# If --force-restart is passed, it will forcefully kill any process listening on the registration port.
# Function to ensure the registration service is running
ensure_registration_service() {
local python_script="$BASE_PATH/registration.py" # Adjust name if needed
local pid_file="$BASE_PATH/registration.pid"
local log_file="$BASE_PATH/registration.log"
local python_script="$BASE_PATH/registration.py"
local pid_file="$BASE_PATH/data/registration.pid"
local log_file="$BASE_PATH/logs/registration.log"
if [ ! -f "$python_script" ]; then
log "ERROR: Python script $python_script not found"
exit 1
fi
# Retrieve the port from the config file (default to 8000 if not specified)
REG_PORT=$(python3 -c "import yaml, sys; print(yaml.safe_load(open('$CONFIG_FILE')).get('port', 8000))")
log "Registration service port from config: $REG_PORT"
@ -175,7 +156,6 @@ ensure_registration_service() {
echo "$NEW_PID" > "$pid_file"
log "Started registration service with PID $NEW_PID"
else
# Check if there is any process already listening on the port
EXISTING_PIDS=$(lsof -ti tcp:"$REG_PORT")
if [ -n "$EXISTING_PIDS" ]; then
log "Registration service already running on port $REG_PORT with PID(s): $EXISTING_PIDS"
@ -224,8 +204,6 @@ if [ "$REFRESH_TOKEN" = true ]; then
refresh_token
fi
restart_container
# Always ensure the registration service is running.
ensure_registration_service
exit 0

1
registration.pid
View file

@ -1 +0,0 @@
3393774

375
registration.py
View file

@ -7,26 +7,34 @@ import smtplib
import httpx
import logging
import ipaddress
import hashlib
import asyncio
import time
from datetime import datetime, timedelta
from email.message import EmailMessage
from typing import List, Dict, Optional, Tuple, Set, Pattern, Union
from fastapi import FastAPI, Request, Form, HTTPException
from fastapi import FastAPI, Request, Form, HTTPException, Depends, Query
from fastapi.responses import HTMLResponse, JSONResponse
from fastapi.templating import Jinja2Templates
from fastapi.staticfiles import StaticFiles
from starlette.middleware.base import BaseHTTPMiddleware
from ipaddress import IPv4Network, IPv4Address
from nio import AsyncClient, RoomMessageText, RoomMessageNotice
# ---------------------------------------------------------
# 1. Load configuration and setup paths
# ---------------------------------------------------------
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
CONFIG_PATH = os.path.join(BASE_DIR, "config.yaml")
CONFIG_DIR = os.path.join(BASE_DIR, "config")
DATA_DIR = os.path.join(BASE_DIR, "data")
LOGS_DIR = os.path.join(BASE_DIR, "logs")
CONFIG_PATH = os.path.join(CONFIG_DIR, "config.yaml")
with open(CONFIG_PATH, "r") as f:
config = yaml.safe_load(f)
# Initialize or load registrations.json
REGISTRATIONS_PATH = os.path.join(BASE_DIR, "registrations.json")
REGISTRATIONS_PATH = os.path.join(DATA_DIR, "registrations.json")
def load_registrations() -> List[Dict]:
try:
with open(REGISTRATIONS_PATH, "r") as f:
@ -34,6 +42,10 @@ def load_registrations() -> List[Dict]:
except (FileNotFoundError, json.JSONDecodeError):
return []
def save_registrations(registrations: List[Dict]):
with open(REGISTRATIONS_PATH, "w") as f:
json.dump(registrations, f, indent=2)
def save_registration(data: Dict):
registrations = load_registrations()
registrations.append(data)
@ -45,7 +57,7 @@ def load_banned_usernames() -> List[Pattern]:
"""Load banned usernames file and compile regex patterns."""
patterns = []
try:
with open(os.path.join(BASE_DIR, "banned_usernames.txt"), "r") as f:
with open(os.path.join(CONFIG_DIR, "banned_usernames.txt"), "r") as f:
for line in f:
line = line.strip()
if line:
@ -62,7 +74,7 @@ def is_ip_banned(ip: str) -> bool:
try:
check_ip = IPv4Address(ip)
try:
with open(os.path.join(BASE_DIR, "banned_ips.txt"), "r") as f:
with open(os.path.join(CONFIG_DIR, "banned_ips.txt"), "r") as f:
for line in f:
line = line.strip()
if not line:
@ -85,13 +97,11 @@ def is_ip_banned(ip: str) -> bool:
def is_email_banned(email: str) -> bool:
"""Check if an email matches any banned patterns."""
try:
with open(os.path.join(BASE_DIR, "banned_emails.txt"), "r") as f:
with open(os.path.join(CONFIG_DIR, "banned_emails.txt"), "r") as f:
for line in f:
pattern = line.strip()
if not pattern:
continue
# Convert email patterns to regex
# Replace * with .* and escape dots
regex_pattern = pattern.replace(".", "\\.").replace("*", ".*")
try:
if re.match(regex_pattern, email, re.IGNORECASE):
@ -107,7 +117,7 @@ def is_username_banned(username: str) -> bool:
patterns = load_banned_usernames()
return any(pattern.search(username) for pattern in patterns)
# Read the registration token
# Read the registration token (still at base level as per shell script)
def read_registration_token():
token_path = os.path.join(BASE_DIR, ".registration_token")
try:
@ -121,7 +131,9 @@ def read_registration_token():
# ---------------------------------------------------------
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
format='%(asctime)s - %(levelname)s - %(message)s',
filename=os.path.join(LOGS_DIR, "registration.log"),
filemode='a'
)
logging.getLogger("uvicorn.access").setLevel(logging.WARNING)
@ -132,7 +144,6 @@ class CustomLoggingMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request, call_next):
if request.url.path == "/api/time" or request.url.path.endswith('favicon.ico'):
return await call_next(request)
response = await call_next(request)
logger.info(f"Request: {request.method} {request.url.path} - Status: {response.status_code}")
return response
@ -145,17 +156,16 @@ def get_current_utc() -> datetime:
def get_next_reset_time(now: datetime) -> datetime:
"""Return the next reset time (possibly today or tomorrow) from config."""
reset_h = config["token_reset_time_utc"] // 100
reset_m = config["token_reset_time_utc"] % 100
reset_h = config["registration"]["token_reset_time_utc"] // 100
reset_m = config["registration"]["token_reset_time_utc"] % 100
candidate = now.replace(hour=reset_h, minute=reset_m, second=0, microsecond=0)
if candidate <= now:
# If we've passed today's reset time, it must be tomorrow.
candidate += timedelta(days=1)
return candidate
def get_downtime_start(next_reset: datetime) -> datetime:
"""Return the downtime start time (minutes before next_reset)."""
return next_reset - timedelta(minutes=config["downtime_before_token_reset"])
return next_reset - timedelta(minutes=config["registration"]["downtime_before_token_reset"])
def format_timedelta(td: timedelta) -> str:
"""Format a timedelta as 'X hours and Y minutes' (or similar)."""
@ -174,7 +184,7 @@ def format_timedelta(td: timedelta) -> str:
elif minutes > 1:
parts.append(f"{minutes} minutes")
if not parts: # If total is less than a minute
if not parts:
return "0 minutes"
return " and ".join(parts)
@ -194,7 +204,6 @@ def is_registration_closed(now: datetime) -> Tuple[bool, str]:
ds = get_downtime_start(nr)
if ds <= now < nr:
# We are within downtime
time_until_open = nr - now
msg = (
f"Registration is closed. "
@ -202,9 +211,7 @@ def is_registration_closed(now: datetime) -> Tuple[bool, str]:
)
return True, msg
else:
# Registration is open
if now > ds:
# We've passed ds, so next downtime is tomorrow
nr += timedelta(days=1)
ds = get_downtime_start(nr)
@ -226,10 +233,10 @@ def check_email_cooldown(email: str) -> Optional[str]:
if not email_entries:
return None
if not config.get("multiple_users_per_email", True):
if not config["registration"].get("multiple_users_per_email", True):
return "This email address has already been used to register an account."
email_cooldown = config.get("email_cooldown")
email_cooldown = config["registration"].get("email_cooldown")
if email_cooldown:
latest_registration = max(
datetime.fromisoformat(e["datetime"])
@ -253,7 +260,7 @@ async def check_username_availability(username: str) -> bool:
logger.info(f"[USERNAME CHECK] {username}: Already requested")
return False
url = f"https://{config['homeserver']}/_matrix/client/v3/register/available?username={username}"
url = f"{config['base_url']}/_matrix/client/v3/register/available?username={username}"
async with httpx.AsyncClient() as client:
try:
response = await client.get(url, timeout=5)
@ -273,21 +280,31 @@ async def check_username_availability(username: str) -> bool:
# ---------------------------------------------------------
# 5. Email Helper Functions
# ---------------------------------------------------------
def load_template(template_path: str) -> str:
"""Load an email template from a file."""
try:
with open(os.path.join(BASE_DIR, template_path), "r") as f:
return f.read()
except FileNotFoundError:
raise HTTPException(status_code=500, detail=f"Email template not found: {template_path}")
def build_email_message(token: str, requested_username: str, now: datetime, recipient_email: str) -> EmailMessage:
"""
Build and return an EmailMessage for registration.
Build and return an EmailMessage for registration using file-based templates.
"""
time_until_reset = get_time_until_reset_str(now)
# Format bodies using config templates
plain_body = config["email_body"].format(
plain_template = load_template(config["email"]["templates"]["registration_token"]["body"])
html_template = load_template(config["email"]["templates"]["registration_token"]["body_html"])
plain_body = plain_template.format(
homeserver=config["homeserver"],
registration_token=token,
requested_username=requested_username,
utc_time=now.strftime("%H:%M:%S"),
time_until_reset=time_until_reset
)
html_body = config.get("email_body_html", "").format(
html_body = html_template.format(
homeserver=config["homeserver"],
registration_token=token,
requested_username=requested_username,
@ -297,21 +314,10 @@ def build_email_message(token: str, requested_username: str, now: datetime, reci
msg = EmailMessage()
msg.set_content(plain_body)
msg.add_alternative(html_body, subtype="html")
if html_body:
msg.add_alternative(html_body, subtype="html")
msg["Subject"] = config["email_subject"].format(homeserver=config["homeserver"])
# Get the sender value from configuration.
# Ensure it's fully-qualified: it must contain an "@".
from_value = config["smtp"].get("from")
if not from_value or "@" not in from_value:
logger.warning(f"Sender address '{from_value}' is not fully-qualified. Falling back to {config['smtp']['username']}.")
from_value = config["smtp"]["username"]
msg["From"] = from_value
msg["Subject"] = config["email"]["templates"]["registration_token"]["subject"].format(homeserver=config["homeserver"])
msg["From"] = config["email"]["smtp"]["from"]
msg["To"] = recipient_email
return msg
@ -319,7 +325,7 @@ def send_email_message(msg: EmailMessage) -> None:
"""
Send an email message using SMTP configuration.
"""
smtp_conf = config["smtp"]
smtp_conf = config["email"]["smtp"]
try:
with smtplib.SMTP(smtp_conf["host"], smtp_conf["port"]) as server:
if smtp_conf.get("use_tls", True):
@ -336,8 +342,133 @@ def send_email_message(msg: EmailMessage) -> None:
# ---------------------------------------------------------
app = FastAPI()
app.add_middleware(CustomLoggingMiddleware)
app.mount("/static", StaticFiles(directory="static"), name="static")
templates = Jinja2Templates(directory="templates")
app.mount("/static", StaticFiles(directory=os.path.join(BASE_DIR, "static")), name="static")
templates = Jinja2Templates(directory=os.path.join(BASE_DIR, "templates"))
# Dependency for admin authentication
def verify_admin_auth(auth_token: str = Form(...)) -> None:
"""Verify the SHA256 hash of matrix_admin.password from config.yaml."""
expected_password = config["matrix_admin"].get("password", "")
expected_hash = hashlib.sha256(expected_password.encode()).hexdigest()
if auth_token != expected_hash:
raise HTTPException(status_code=403, detail="Invalid authentication token")
# Helper function to parse Matrix responses
def parse_response(response_text: str, query: str) -> Dict[str, Union[str, List[str]]]:
"""Parse a response that may contain a markdown codeblock."""
query_parts = query.strip().split()
array_key = query_parts[0] if query_parts else "data"
codeblock_pattern = r"(.*?):\s*\n```\s*\n([\s\S]*?)\n```"
match = re.search(codeblock_pattern, response_text)
if match:
message = match.group(1).strip()
codeblock_content = match.group(2)
items = [line for line in codeblock_content.split('\n') if line.strip()]
return {"message": message, array_key: items}
return {"response": response_text}
# Helper function to get the list of users from the Matrix admin room
async def get_matrix_users() -> List[str]:
"""Fetch the list of users from the Matrix admin room."""
matrix_config = config["matrix_admin"]
homeserver = config["base_url"]
username = matrix_config.get("username")
password = matrix_config.get("password")
admin_room = matrix_config.get("room")
admin_response_user = matrix_config.get("super_admin")
if not all([homeserver, username, password, admin_room, admin_response_user]):
raise HTTPException(status_code=500, detail="Incomplete Matrix admin configuration")
client = AsyncClient(homeserver, username)
try:
login_response = await client.login(password)
if getattr(login_response, "error", None):
raise Exception(f"Login error: {login_response.error}")
logger.debug("Successfully logged in to Matrix")
await client.join(admin_room)
initial_sync = await client.sync(timeout=5000)
next_batch = initial_sync.next_batch
await client.room_send(
room_id=admin_room,
message_type="m.room.message",
content={"msgtype": "m.text", "body": "!admin users list-users"},
)
query_time = time.time()
timeout_seconds = 10
start_time = time.time()
response_message = None
while (time.time() - start_time) < timeout_seconds:
sync_response = await client.sync(timeout=2000, since=next_batch)
next_batch = sync_response.next_batch
room = sync_response.rooms.join.get(admin_room)
if room and room.timeline and room.timeline.events:
message_events = [
event for event in room.timeline.events
if isinstance(event, (RoomMessageText, RoomMessageNotice))
]
for event in message_events:
event_time = event.server_timestamp / 1000.0
if event.sender == admin_response_user and event_time >= query_time:
response_message = event.body
logger.debug(f"Found response: {response_message[:100]}...")
break
if response_message:
break
await client.logout()
await client.close()
if not response_message:
raise HTTPException(status_code=504, detail="No response from admin user within timeout")
parsed = parse_response(response_message, "users list-users")
return parsed.get("users", [])
except Exception as e:
await client.close()
logger.error(f"Error fetching Matrix users: {e}")
raise HTTPException(status_code=500, detail=f"Error fetching users: {e}")
# Helper function to deactivate a user via Matrix admin room
async def deactivate_user(user: str) -> bool:
"""Send a deactivation command for a user to the Matrix admin room."""
matrix_config = config["matrix_admin"]
homeserver = config["base_url"]
username = matrix_config.get("username")
password = matrix_config.get("password")
admin_room = matrix_config.get("room")
admin_response_user = matrix_config.get("super_admin")
client = AsyncClient(homeserver, username)
try:
login_response = await client.login(password)
if getattr(login_response, "error", None):
raise Exception(f"Login error: {login_response.error}")
logger.debug(f"Logged in to deactivate {user}")
await client.join(admin_room)
await client.sync(timeout=5000)
command = f"!admin users deactivate {user}"
await client.room_send(
room_id=admin_room,
message_type="m.room.message",
content={"msgtype": "m.text", "body": command},
)
logger.info(f"Sent deactivation command for {user}")
await asyncio.sleep(1)
await client.logout()
await client.close()
return True
except Exception as e:
await client.close()
logger.error(f"Failed to deactivate {user}: {e}")
return False
@app.get("/", response_class=HTMLResponse)
async def index(request: Request):
@ -351,9 +482,9 @@ async def index(request: Request):
"registration_closed": closed,
"homeserver": config["homeserver"],
"message": message,
"reset_hour": config["token_reset_time_utc"] // 100,
"reset_minute": config["token_reset_time_utc"] % 100,
"downtime_minutes": config["downtime_before_token_reset"]
"reset_hour": config["registration"]["token_reset_time_utc"] // 100,
"reset_minute": config["registration"]["token_reset_time_utc"] % 100,
"downtime_minutes": config["registration"]["downtime_before_token_reset"]
}
)
@ -400,11 +531,9 @@ async def register(
logger.error("Registration token file not found")
raise HTTPException(status_code=500, detail="Registration token file not found.")
# Build and send registration email
email_message = build_email_message(token, requested_username, now, email)
send_email_message(email_message)
# Save registration data and log success
registration_data = {
"requested_name": requested_username,
"email": email,
@ -416,6 +545,156 @@ async def register(
return templates.TemplateResponse("success.html", {"request": request, "homeserver": config["homeserver"]})
@app.post("/_admin/purge_unfulfilled_registrations", response_class=JSONResponse)
async def purge_unfulfilled_registrations(
min_age_hours: int = Form(default=24),
auth_token: str = Depends(verify_admin_auth)
):
"""
Purge unfulfilled registration entries older than min_age_hours where the username
does not exist on the homeserver.
"""
registrations = load_registrations()
if not registrations:
return JSONResponse({"message": "No registrations found to clean up"})
logger.info(f"Starting cleanup of {len(registrations)} registrations")
logger.info(f"Will remove non-existent users registered more than {min_age_hours} hours ago")
entries_to_keep = []
removed_count = 0
too_new_count = 0
exists_count = 0
current_time = datetime.utcnow()
async with httpx.AsyncClient() as client:
for entry in registrations:
username = entry["requested_name"]
reg_date = datetime.fromisoformat(entry["datetime"])
age = current_time - reg_date
url = f"{config['base_url']}/_matrix/client/v3/register/available?username={username}"
try:
response = await client.get(url, timeout=5)
if response.status_code == 200 and response.json().get("available", False):
exists = False
elif response.status_code == 400 or (response.status_code == 200 and not response.json().get("available", False)):
exists = True
else:
logger.warning(f"Unexpected response for {username}: {response.status_code}")
exists = False
except httpx.RequestError as ex:
logger.error(f"Error checking username {username}: {ex}")
exists = False
if exists:
entries_to_keep.append(entry)
exists_count += 1
logger.info(f"Keeping registration for existing user: {username}")
continue
if age < timedelta(hours=min_age_hours):
entries_to_keep.append(entry)
too_new_count += 1
logger.info(f"Keeping recent registration: {username} (age: {age.total_seconds()/3600:.1f} hours)")
else:
logger.info(f"Removing old registration: {username} (age: {age.total_seconds()/3600:.1f} hours)")
removed_count += 1
save_registrations(entries_to_keep)
result = {
"message": "Cleanup complete",
"kept_existing": exists_count,
"kept_recent": too_new_count,
"removed": removed_count,
"total_remaining": len(entries_to_keep)
}
logger.info(f"Cleanup complete: {result}")
return JSONResponse(result)
@app.post("/_admin/deactivate_undocumented_users", response_class=JSONResponse)
async def deactivate_undocumented_users(auth_token: str = Depends(verify_admin_auth)):
"""Deactivate users on the homeserver without matching entries in registrations.json."""
registrations = load_registrations()
matrix_users = await get_matrix_users()
registered_usernames = {entry["requested_name"].lower() for entry in registrations}
homeserver = config["homeserver"].lower()
undocumented_users = []
for user in matrix_users:
if not user.lower().startswith("@"):
continue
username, user_homeserver = user[1:].lower().split(":", 1)
if user_homeserver != homeserver:
continue
if username not in registered_usernames:
undocumented_users.append(user)
if not undocumented_users:
logger.info("No undocumented users found to deactivate")
return JSONResponse({"message": "No undocumented users found to deactivate", "deactivated_count": 0})
deactivated_count = 0
failed_deactivations = []
for user in undocumented_users:
success = await deactivate_user(user)
if success:
deactivated_count += 1
else:
failed_deactivations.append(user)
logger.info(f"Deactivated {deactivated_count} undocumented users")
if failed_deactivations:
logger.warning(f"Failed to deactivate {len(failed_deactivations)} users: {failed_deactivations}")
result = {
"message": f"Deactivated {deactivated_count} undocumented user(s)",
"deactivated_count": deactivated_count
}
if failed_deactivations:
result["failed_deactivations"] = failed_deactivations
return JSONResponse(result)
@app.post("/_admin/retroactively_document_users", response_class=JSONResponse)
async def retroactively_document_users(auth_token: str = Depends(verify_admin_auth)):
"""Add entries to registrations.json for undocumented users."""
registrations = load_registrations()
matrix_users = await get_matrix_users()
registered_usernames = {entry["requested_name"].lower() for entry in registrations}
homeserver = config["homeserver"].lower()
added_count = 0
for user in matrix_users:
if not user.lower().startswith("@"):
continue
username, user_homeserver = user[1:].lower().split(":", 1)
if user_homeserver != homeserver:
continue
if username not in registered_usernames:
new_entry = {
"requested_name": username,
"email": "null@nope.no",
"datetime": datetime.utcnow().isoformat(),
"ip_address": "127.0.0.1"
}
registrations.append(new_entry)
registered_usernames.add(username)
added_count += 1
logger.info(f"Added retroactive entry for {user}")
if added_count > 0:
save_registrations(registrations)
logger.info(f"Retroactively documented {added_count} users")
return JSONResponse({
"message": f"Retroactively documented {added_count} user(s)",
"added_count": added_count
})
if __name__ == "__main__":
import uvicorn
uvicorn.run(

130
sw1tch/__init__.py Normal file
View file

@ -0,0 +1,130 @@
import os
import yaml
import json
import logging
import re
import hashlib
from typing import List, Dict, Pattern
from fastapi import HTTPException
from starlette.middleware.base import BaseHTTPMiddleware
from ipaddress import IPv4Address, IPv4Network
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
CONFIG_DIR = os.path.join(BASE_DIR, "config")
DATA_DIR = os.path.join(BASE_DIR, "data")
LOGS_DIR = os.path.join(BASE_DIR, "logs")
CONFIG_PATH = os.path.join(CONFIG_DIR, "config.yaml")
with open(CONFIG_PATH, "r") as f:
config = yaml.safe_load(f)
REGISTRATIONS_PATH = os.path.join(DATA_DIR, "registrations.json")
def load_registrations() -> List[Dict]:
try:
with open(REGISTRATIONS_PATH, "r") as f:
return json.load(f)
except (FileNotFoundError, json.JSONDecodeError):
return []
def save_registrations(registrations: List[Dict]):
with open(REGISTRATIONS_PATH, "w") as f:
json.dump(registrations, f, indent=2)
def save_registration(data: Dict):
registrations = load_registrations()
registrations.append(data)
save_registrations(registrations)
def load_banned_usernames() -> List[Pattern]:
patterns = []
try:
with open(os.path.join(CONFIG_DIR, "banned_usernames.txt"), "r") as f:
for line in f:
line = line.strip()
if line:
try:
patterns.append(re.compile(line, re.IGNORECASE))
except re.error:
logging.error(f"Invalid regex pattern in banned_usernames.txt: {line}")
except FileNotFoundError:
pass
return patterns
def is_ip_banned(ip: str) -> bool:
try:
check_ip = IPv4Address(ip)
try:
with open(os.path.join(CONFIG_DIR, "banned_ips.txt"), "r") as f:
for line in f:
line = line.strip()
if not line:
continue
try:
if '/' in line:
if check_ip in IPv4Network(line):
return True
else:
if check_ip == IPv4Address(line):
return True
except ValueError:
logging.error(f"Invalid IP/CIDR in banned_ips.txt: {line}")
except FileNotFoundError:
return False
except ValueError:
logging.error(f"Invalid IP address to check: {ip}")
return False
def is_email_banned(email: str) -> bool:
try:
with open(os.path.join(CONFIG_DIR, "banned_emails.txt"), "r") as f:
for line in f:
pattern = line.strip()
if not pattern:
continue
regex_pattern = pattern.replace(".", "\\.").replace("*", ".*")
try:
if re.match(regex_pattern, email, re.IGNORECASE):
return True
except re.error:
logging.error(f"Invalid email pattern in banned_emails.txt: {pattern}")
except FileNotFoundError:
pass
return False
def is_username_banned(username: str) -> bool:
patterns = load_banned_usernames()
return any(pattern.search(username) for pattern in patterns)
def read_registration_token():
token_path = os.path.join(DATA_DIR, ".registration_token")
try:
with open(token_path, "r") as f:
return f.read().strip()
except FileNotFoundError:
return None
# Logging setup
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
filename=os.path.join(LOGS_DIR, "registration.log"),
filemode='a'
)
logging.getLogger("uvicorn.access").setLevel(logging.WARNING)
logging.getLogger("uvicorn.error").setLevel(logging.WARNING)
logger = logging.getLogger(__name__)
class CustomLoggingMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request, call_next):
if request.url.path == "/api/time" or request.url.path.endswith('favicon.ico'):
return await call_next(request)
response = await call_next(request)
logger.info(f"Request: {request.method} {request.url.path} - Status: {response.status_code}")
return response
def verify_admin_auth(auth_token: str) -> None:
expected_password = config["matrix_admin"].get("password", "")
expected_hash = hashlib.sha256(expected_password.encode()).hexdigest()
if auth_token != expected_hash:
raise HTTPException(status_code=403, detail="Invalid authentication token")

25
sw1tch/__main__.py Normal file
View file

@ -0,0 +1,25 @@
import os
from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles
from sw1tch import BASE_DIR, CustomLoggingMiddleware
from sw1tch.routes.public import router as public_router
from sw1tch.routes.admin import router as admin_router
from sw1tch.routes.canary import router as canary_router
app = FastAPI()
app.add_middleware(CustomLoggingMiddleware)
app.mount("/static", StaticFiles(directory=os.path.join(BASE_DIR, "static")), name="static")
app.include_router(public_router)
app.include_router(admin_router)
app.include_router(canary_router)
if __name__ == "__main__":
import uvicorn
from sw1tch import config
uvicorn.run(
"sw1tch.__main__:app", # import string format required for reload
host="0.0.0.0",
port=config["port"],
reload=True,
access_log=False
)

0
canary.py → sw1tch/canary.py
View file

124
sw1tch/routes/admin.py Normal file
View file

@ -0,0 +1,124 @@
from fastapi import APIRouter, Form, Depends
from fastapi.responses import JSONResponse
from datetime import datetime, timedelta
import httpx
import re
from sw1tch import config, logger, load_registrations, save_registrations, verify_admin_auth
from sw1tch.utilities.matrix import get_matrix_users, deactivate_user
router = APIRouter(prefix="/_admin", dependencies=[Depends(verify_admin_auth)])
@router.post("/purge_unfulfilled_registrations", response_class=JSONResponse)
async def purge_unfulfilled_registrations(min_age_hours: int = Form(default=24)):
registrations = load_registrations()
if not registrations:
return JSONResponse({"message": "No registrations found to clean up"})
logger.info(f"Starting cleanup of {len(registrations)} registrations")
logger.info(f"Will remove non-existent users registered more than {min_age_hours} hours ago")
entries_to_keep = []
removed_count = 0
too_new_count = 0
exists_count = 0
current_time = datetime.utcnow()
async with httpx.AsyncClient() as client:
for entry in registrations:
username = entry["requested_name"]
reg_date = datetime.fromisoformat(entry["datetime"])
age = current_time - reg_date
url = f"{config['base_url']}/_matrix/client/v3/register/available?username={username}"
try:
response = await client.get(url, timeout=5)
if response.status_code == 200 and response.json().get("available", False):
exists = False
elif response.status_code == 400 or (response.status_code == 200 and not response.json().get("available", False)):
exists = True
else:
logger.warning(f"Unexpected response for {username}: {response.status_code}")
exists = False
except httpx.RequestError as ex:
logger.error(f"Error checking username {username}: {ex}")
exists = False
if exists:
entries_to_keep.append(entry)
exists_count += 1
logger.info(f"Keeping registration for existing user: {username}")
continue
if age < timedelta(hours=min_age_hours):
entries_to_keep.append(entry)
too_new_count += 1
logger.info(f"Keeping recent registration: {username} (age: {age.total_seconds()/3600:.1f} hours)")
else:
logger.info(f"Removing old registration: {username} (age: {age.total_seconds()/3600:.1f} hours)")
removed_count += 1
save_registrations(entries_to_keep)
result = {
"message": "Cleanup complete",
"kept_existing": exists_count,
"kept_recent": too_new_count,
"removed": removed_count,
"total_remaining": len(entries_to_keep)
}
logger.info(f"Cleanup complete: {result}")
return JSONResponse(result)
@router.post("/deactivate_undocumented_users", response_class=JSONResponse)
async def deactivate_undocumented_users():
registrations = load_registrations()
matrix_users = await get_matrix_users()
registered_usernames = {entry["requested_name"].lower() for entry in registrations}
homeserver = config["homeserver"].lower()
undocumented_users = [
user for user in matrix_users
if user.lower().startswith("@") and user[1:].lower().split(":", 1)[1] == homeserver
and user[1:].lower().split(":", 1)[0] not in registered_usernames
]
if not undocumented_users:
logger.info("No undocumented users found to deactivate")
return JSONResponse({"message": "No undocumented users found to deactivate", "deactivated_count": 0})
deactivated_count = 0
failed_deactivations = []
for user in undocumented_users:
success = await deactivate_user(user)
if success:
deactivated_count += 1
else:
failed_deactivations.append(user)
logger.info(f"Deactivated {deactivated_count} undocumented users")
if failed_deactivations:
logger.warning(f"Failed to deactivate {len(failed_deactivations)} users: {failed_deactivations}")
result = {"message": f"Deactivated {deactivated_count} undocumented user(s)", "deactivated_count": deactivated_count}
if failed_deactivations:
result["failed_deactivations"] = failed_deactivations
return JSONResponse(result)
@router.post("/retroactively_document_users", response_class=JSONResponse)
async def retroactively_document_users():
registrations = load_registrations()
matrix_users = await get_matrix_users()
registered_usernames = {entry["requested_name"].lower() for entry in registrations}
homeserver = config["homeserver"].lower()
added_count = 0
for user in matrix_users:
if not user.lower().startswith("@"):
continue
username, user_homeserver = user[1:].lower().split(":", 1)
if user_homeserver != homeserver or username in registered_usernames:
continue
new_entry = {
"requested_name": username,
"email": "null@nope.no",
"datetime": datetime.utcnow().isoformat(),
"ip_address": "127.0.0.1"
}
registrations.append(new_entry)
registered_usernames.add(username)
added_count += 1
logger.info(f"Added retroactive entry for {user}")
if added_count > 0:
save_registrations(registrations)
logger.info(f"Retroactively documented {added_count} users")
return JSONResponse({
"message": f"Retroactively documented {added_count} user(s)",
"added_count": added_count
})

167
sw1tch/routes/canary.py Normal file
View file

@ -0,0 +1,167 @@
import os
import subprocess
import requests
import feedparser
import datetime
from typing import List
from fastapi import APIRouter, Request, Form, Depends, HTTPException
from fastapi.responses import HTMLResponse, JSONResponse
from fastapi.templating import Jinja2Templates
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
from sw1tch import BASE_DIR, config, logger, verify_admin_auth
from sw1tch.utilities.matrix import AsyncClient
router = APIRouter(prefix="/_admin/warrant_canary", dependencies=[Depends(verify_admin_auth)])
templates = Jinja2Templates(directory=os.path.join(BASE_DIR, "templates"))
ATTESTATIONS_FILE = os.path.join(BASE_DIR, "config", "attestations.txt")
CANARY_OUTPUT_FILE = os.path.join(BASE_DIR, "data", "canary.txt")
TEMP_CANARY_FILE = os.path.join(BASE_DIR, "data", "temp_canary_message.txt")
def load_attestations():
try:
with open(ATTESTATIONS_FILE, 'r') as f:
return [line.strip() for line in f if line.strip()]
except FileNotFoundError:
raise HTTPException(status_code=500, detail=f"Attestations file not found: {ATTESTATIONS_FILE}")
def get_nist_time():
session = requests.Session()
retry_strategy = Retry(total=3, backoff_factor=1, status_forcelist=[500, 502, 503, 504])
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("https://", adapter)
endpoints = [
"https://timeapi.io/api/Time/current/zone?timeZone=UTC",
"https://worldtimeapi.org/api/timezone/UTC",
]
for url in endpoints:
try:
response = session.get(url, timeout=10)
response.raise_for_status()
data = response.json()
if "dateTime" in data:
return data["dateTime"] + " UTC"
elif "utc_datetime" in data:
return data["utc_datetime"] + " UTC"
except requests.RequestException:
pass
raise HTTPException(status_code=500, detail="Failed to fetch NIST time")
def get_rss_headline():
rss_config = config['canary'].get('rss', {})
rss_url = rss_config.get('url', 'https://www.democracynow.org/democracynow.rss')
feed = feedparser.parse(rss_url)
if feed.entries and len(feed.entries) > 0:
return {"title": feed.entries[0].title, "link": feed.entries[0].link}
raise HTTPException(status_code=500, detail="Failed to fetch RSS headline")
def get_bitcoin_latest_block():
try:
response = requests.get("https://blockchain.info/latestblock", timeout=10)
if response.status_code == 200:
data = response.json()
block_response = requests.get(f"https://blockchain.info/rawblock/{data['hash']}", timeout=10)
if block_response.status_code == 200:
block_data = block_response.json()
hash_str = data["hash"].lstrip("0") or "0"
return {
"height": data["height"],
"hash": hash_str,
"time": datetime.datetime.fromtimestamp(block_data["time"]).strftime("%Y-%m-%d %H:%M:%S UTC")
}
except Exception:
raise HTTPException(status_code=500, detail="Failed to fetch Bitcoin block data")
def create_warrant_canary_message(attestations: List[str], note: str):
nist_time = get_nist_time()
rss_data = get_rss_headline()
bitcoin_block = get_bitcoin_latest_block()
org = config['canary']['organization']
admin_name = config['canary'].get('admin_name', 'Admin')
admin_title = config['canary'].get('admin_title', 'administrator')
message = f"{org} Warrant Canary · {nist_time}\n"
message += f"I, {admin_name}, the {admin_title} of {org}, state this {datetime.datetime.now().strftime('%dth day of %B, %Y')}:\n"
for i, attestation in enumerate(attestations, 1):
message += f" {i}. {org} {attestation}\n"
if note:
message += f"\nNOTE: {note}\n"
message += "\nDatestamp Proof:\n"
message += f" Daily News: \"{rss_data['title']}\"\n"
message += f" Source URL: {rss_data['link']}\n"
message += f" BTC block: #{bitcoin_block['height']}, {bitcoin_block['time']}\n"
message += f" Block hash: {bitcoin_block['hash']}\n"
return message.rstrip() + "\n"
def sign_with_gpg(message: str, gpg_key_id: str, passphrase: str):
try:
with open(TEMP_CANARY_FILE, "w", newline='\n') as f:
f.write(message)
cmd = ["gpg", "--batch", "--yes", "--passphrase", passphrase, "--clearsign", "--default-key", gpg_key_id, TEMP_CANARY_FILE]
subprocess.run(cmd, check=True)
with open(f"{TEMP_CANARY_FILE}.asc", "r") as f:
signed_message = f.read()
os.remove(TEMP_CANARY_FILE)
os.remove(f"{TEMP_CANARY_FILE}.asc")
lines = signed_message.splitlines()
signature_idx = next(i for i, line in enumerate(lines) if line == "-----BEGIN PGP SIGNATURE-----")
if lines[signature_idx + 1] == "":
lines.pop(signature_idx + 1)
return "\n".join(lines)
except subprocess.CalledProcessError as e:
raise HTTPException(status_code=500, detail=f"GPG signing failed: {e}")
async def post_to_matrix(signed_message: str):
try:
matrix = config['canary']['credentials']
client = AsyncClient(config['base_url'], matrix['username'])
await client.login(matrix['password'])
full_message = (
f"This is the {config['canary']['organization']} Warrant Canary, signed with GPG for authenticity. "
"Copy the code block below to verify with `gpg --verify`:\n\n"
f"```\n{signed_message}\n```"
)
content = {
"msgtype": "m.text",
"body": full_message,
"format": "org.matrix.custom.html",
"formatted_body": (
f"This is the {config['canary']['organization']} Warrant Canary, signed with GPG for authenticity. "
"Copy the code block below to verify with <code>gpg --verify</code>:<br><br>"
f"<pre>{signed_message}</pre>"
)
}
await client.room_send(config['matrix_admin']['room'], "m.room.message", content)
await client.logout()
await client.close()
return True
except Exception as e:
logger.error(f"Error posting to Matrix: {e}")
return False
@router.get("/", response_class=HTMLResponse)
async def warrant_canary_form(request: Request):
attestations = load_attestations()
return templates.TemplateResponse("canary_form.html", {
"request": request,
"attestations": attestations,
"organization": config["canary"]["organization"]
})
@router.post("/preview", response_class=HTMLResponse)
async def warrant_canary_preview(request: Request, attestations: List[str] = Form(...), note: str = Form(default="")):
message = create_warrant_canary_message(attestations, note)
return templates.TemplateResponse("canary_preview.html", {"request": request, "message": message})
@router.post("/sign", response_class=HTMLResponse)
async def warrant_canary_sign(request: Request, message: str = Form(...), passphrase: str = Form(...)):
signed_message = sign_with_gpg(message, config["canary"]["gpg_key_id"], passphrase)
with open(CANARY_OUTPUT_FILE, "w") as f:
f.write(signed_message)
return templates.TemplateResponse("canary_success.html", {"request": request, "signed_message": signed_message})
@router.post("/post", response_class=JSONResponse)
async def warrant_canary_post(signed_message: str = Form(...)):
success = await post_to_matrix(signed_message)
return JSONResponse({"message": "Posted to Matrix" if success else "Failed to post to Matrix"})

73
sw1tch/routes/public.py Normal file
View file

@ -0,0 +1,73 @@
import os
from datetime import datetime as datetime
from fastapi import APIRouter, Request, Form
from fastapi.responses import HTMLResponse, JSONResponse
from fastapi.templating import Jinja2Templates
from sw1tch import BASE_DIR, config, logger, read_registration_token
from sw1tch.utilities.time import get_current_utc, is_registration_closed
from sw1tch.utilities.registration import check_email_cooldown, check_username_availability, build_email_message, send_email_message
from sw1tch import save_registration, is_ip_banned, is_email_banned
router = APIRouter()
templates = Jinja2Templates(directory=os.path.join(BASE_DIR, "templates"))
@router.get("/", response_class=HTMLResponse)
async def index(request: Request):
now = get_current_utc()
closed, message = is_registration_closed(now)
return templates.TemplateResponse(
"index.html",
{
"request": request,
"registration_closed": closed,
"homeserver": config["homeserver"],
"message": message,
"reset_hour": config["registration"]["token_reset_time_utc"] // 100,
"reset_minute": config["registration"]["token_reset_time_utc"] % 100,
"downtime_minutes": config["registration"]["downtime_before_token_reset"]
}
)
@router.get("/api/time")
async def get_server_time():
now = get_current_utc()
return JSONResponse({"utc_time": now.strftime("%H:%M:%S")})
@router.post("/register", response_class=HTMLResponse)
async def register(request: Request, requested_username: str = Form(...), email: str = Form(...)):
now = get_current_utc()
client_ip = request.client.host
logger.info(f"Registration attempt - Username: {requested_username}, Email: {email}, IP: {client_ip}")
closed, message = is_registration_closed(now)
if closed:
logger.info("Registration rejected: Registration is closed")
return templates.TemplateResponse("error.html", {"request": request, "message": message})
if is_ip_banned(client_ip):
logger.info(f"Registration rejected: Banned IP {client_ip}")
return templates.TemplateResponse("error.html", {"request": request, "message": "Registration not allowed from your IP address."})
if is_email_banned(email):
logger.info(f"Registration rejected: Banned email {email}")
return templates.TemplateResponse("error.html", {"request": request, "message": "Registration not allowed for this email address."})
if error_message := check_email_cooldown(email):
logger.info(f"Registration rejected: Email cooldown - {email}")
return templates.TemplateResponse("error.html", {"request": request, "message": error_message})
available = await check_username_availability(requested_username)
if not available:
logger.info(f"Registration rejected: Username unavailable - {requested_username}")
return templates.TemplateResponse("error.html", {"request": request, "message": f"The username '{requested_username}' is not available."})
token = read_registration_token()
if token is None:
logger.error("Registration token file not found")
raise HTTPException(status_code=500, detail="Registration token file not found.")
email_message = build_email_message(token, requested_username, now, email)
send_email_message(email_message)
registration_data = {
"requested_name": requested_username,
"email": email,
"datetime": datetime.utcnow().isoformat(),
"ip_address": client_ip
}
save_registration(registration_data)
logger.info(f"Registration successful - Username: {requested_username}, Email: {email}")
return templates.TemplateResponse("success.html", {"request": request, "homeserver": config["homeserver"]})

0
static/favicon.ico → sw1tch/static/favicon.ico
View file

Side by side Swipe Overlay

Before

Width: 256px  |  Height: 256px  |  Size: 175 KiB

After

Width: 256px  |  Height: 256px  |  Size: 175 KiB

0
static/logo.png → sw1tch/static/logo.png
View file

Side by side Swipe

Before

(image error) Size: 43 KiB

After

(image error) Size: 43 KiB

0
static/styles.css → sw1tch/static/styles.css
View file

17
sw1tch/templates/canary_form.html Normal file
View file

@ -0,0 +1,17 @@
<!DOCTYPE html>
<html>
<head>
<title>Warrant Canary Form</title>
</head>
<body>
<h1>Create Warrant Canary</h1>
<form method="post" action="/_admin/warrant_canary/preview">
{% for attestation in attestations %}
<input type="checkbox" name="attestations" value="{{ attestation }}"> {{ organization }} {{ attestation }}<br>
{% endfor %}
<textarea name="note" placeholder="Optional note"></textarea><br>
<input type="hidden" name="auth_token" value="{{ request.query_params.auth_token }}">
<button type="submit">Preview</button>
</form>
</body>
</html>

16
sw1tch/templates/canary_preview.html Normal file
View file

@ -0,0 +1,16 @@
<!DOCTYPE html>
<html>
<head>
<title>Warrant Canary Preview</title>
</head>
<body>
<h1>Warrant Canary Preview</h1>
<pre>{{ message }}</pre>
<form method="post" action="/_admin/warrant_canary/sign">
<input type="hidden" name="message" value="{{ message|escape }}">
<input type="password" name="passphrase" placeholder="GPG Passphrase" required>
<input type="hidden" name="auth_token" value="{{ request.query_params.auth_token }}">
<button type="submit">Sign</button>
</form>
</body>
</html>

15
sw1tch/templates/canary_success.html Normal file
View file

@ -0,0 +1,15 @@
<!DOCTYPE html>
<html>
<head>
<title>Warrant Canary Signed</title>
</head>
<body>
<h1>Warrant Canary Signed</h1>
<pre>{{ signed_message }}</pre>
<form method="post" action="/_admin/warrant_canary/post">
<input type="hidden" name="signed_message" value="{{ signed_message|escape }}">
<input type="hidden" name="auth_token" value="{{ request.query_params.auth_token }}">
<button type="submit">Post to Matrix</button>
</form>
</body>
</html>

0
sw1tch/templates/emails/account_deactivation.html Normal file
View file

0
sw1tch/templates/emails/account_deactivation.txt Normal file
View file

14
sw1tch/templates/emails/registration_token.html Normal file
View file

@ -0,0 +1,14 @@
<html>
<body style="font-family: Arial, sans-serif; line-height: 1.6;">
<p>Hello,</p>
<p>Thank you for your interest in {homeserver}, {requested_username}.</p>
<p>The registration token today is:</p>
<div style="font-size: 24px; font-weight: bold; padding: 15px; background-color: #f0f0f0; border: 1px solid #ddd; margin: 15px 0; text-align: center;">
{registration_token}
</div>
<p>This registration token is valid for {time_until_reset}. If you do not register in that period, you will need to request the new registration token.</p>
<p>Please ensure you use the username <strong>{requested_username}</strong> when you register. Using a different username may result in your account being deleted at a later time without forewarning.</p>
<p>Regards,<br>
{homeserver} registration team</p>
</body>
</html>

12
sw1tch/templates/emails/registration_token.txt Normal file
View file

@ -0,0 +1,12 @@
Hello,
Thank you for your interest in {homeserver}, {requested_username}.
The registration token today is: {registration_token}
This registration token is valid for {time_until_reset}. If you do not register in that period, you will need to request the new registration token.
Please ensure you use the username {requested_username} when you register. Using a different username may result in your account being deleted at a later time without forewarning.
Regards,
{homeserver} registration team

0
templates/error.html → sw1tch/templates/error.html
View file

0
templates/index.html → sw1tch/templates/index.html
View file

0
templates/success.html → sw1tch/templates/success.html
View file

101
sw1tch/utilities/matrix.py Normal file
View file

@ -0,0 +1,101 @@
import asyncio
import time
import re
from typing import List, Dict, Union
from fastapi import HTTPException
from nio import AsyncClient, RoomMessageText, RoomMessageNotice
from sw1tch import config, logger
def parse_response(response_text: str, query: str) -> Dict[str, Union[str, List[str]]]:
query_parts = query.strip().split()
array_key = query_parts[0] if query_parts else "data"
codeblock_pattern = r"(.*?):\s*\n```\s*\n([\s\S]*?)\n```"
match = re.search(codeblock_pattern, response_text)
if match:
message = match.group(1).strip()
items = [line for line in match.group(2).split('\n') if line.strip()]
return {"message": message, array_key: items}
return {"response": response_text}
async def get_matrix_users() -> List[str]:
matrix_config = config["matrix_admin"]
homeserver = config["base_url"]
username = matrix_config.get("username")
password = matrix_config.get("password")
admin_room = matrix_config.get("room")
admin_response_user = matrix_config.get("super_admin")
if not all([homeserver, username, password, admin_room, admin_response_user]):
raise HTTPException(status_code=500, detail="Incomplete Matrix admin configuration")
client = AsyncClient(homeserver, username)
try:
login_response = await client.login(password)
if getattr(login_response, "error", None):
raise Exception(f"Login error: {login_response.error}")
logger.debug("Successfully logged in to Matrix")
await client.join(admin_room)
initial_sync = await client.sync(timeout=5000)
next_batch = initial_sync.next_batch
await client.room_send(
room_id=admin_room,
message_type="m.room.message",
content={"msgtype": "m.text", "body": "!admin users list-users"},
)
query_time = time.time()
timeout_seconds = 10
start_time = time.time()
response_message = None
while (time.time() - start_time) < timeout_seconds:
sync_response = await client.sync(timeout=2000, since=next_batch)
next_batch = sync_response.next_batch
room = sync_response.rooms.join.get(admin_room)
if room and room.timeline and room.timeline.events:
message_events = [e for e in room.timeline.events if isinstance(e, (RoomMessageText, RoomMessageNotice))]
for event in message_events:
event_time = event.server_timestamp / 1000.0
if event.sender == admin_response_user and event_time >= query_time:
response_message = event.body
logger.debug(f"Found response: {response_message[:100]}...")
break
if response_message:
break
await client.logout()
await client.close()
if not response_message:
raise HTTPException(status_code=504, detail="No response from admin user within timeout")
parsed = parse_response(response_message, "users list-users")
return parsed.get("users", [])
except Exception as e:
await client.close()
logger.error(f"Error fetching Matrix users: {e}")
raise HTTPException(status_code=500, detail=f"Error fetching users: {e}")
async def deactivate_user(user: str) -> bool:
matrix_config = config["matrix_admin"]
homeserver = config["base_url"]
username = matrix_config.get("username")
password = matrix_config.get("password")
admin_room = matrix_config.get("room")
client = AsyncClient(homeserver, username)
try:
login_response = await client.login(password)
if getattr(login_response, "error", None):
raise Exception(f"Login error: {login_response.error}")
logger.debug(f"Logged in to deactivate {user}")
await client.join(admin_room)
await client.sync(timeout=5000)
command = f"!admin users deactivate {user}"
await client.room_send(
room_id=admin_room,
message_type="m.room.message",
content={"msgtype": "m.text", "body": command},
)
logger.info(f"Sent deactivation command for {user}")
await asyncio.sleep(1)
await client.logout()
await client.close()
return True
except Exception as e:
await client.close()
logger.error(f"Failed to deactivate {user}: {e}")
return False

92
sw1tch/utilities/registration.py Normal file
View file

@ -0,0 +1,92 @@
import os
import smtplib
import httpx
from datetime import datetime
from email.message import EmailMessage
from typing import Optional
from fastapi import HTTPException
from sw1tch import config, BASE_DIR, load_registrations, save_registration, is_username_banned, logger
async def check_username_availability(username: str) -> bool:
if is_username_banned(username):
logger.info(f"[USERNAME CHECK] {username}: Banned by pattern")
return False
registrations = load_registrations()
if any(r["requested_name"] == username for r in registrations):
logger.info(f"[USERNAME CHECK] {username}: Already requested")
return False
url = f"{config['base_url']}/_matrix/client/v3/register/available?username={username}"
async with httpx.AsyncClient() as client:
try:
response = await client.get(url, timeout=5)
if response.status_code == 200:
is_available = response.json().get("available", False)
logger.info(f"[USERNAME CHECK] {username}: {'Available' if is_available else 'Taken'}")
return is_available
elif response.status_code == 400:
logger.info(f"[USERNAME CHECK] {username}: Taken (400)")
return False
except httpx.RequestError as ex:
logger.warning(f"[USERNAME CHECK] Could not reach homeserver: {ex}")
return False
return False
def check_email_cooldown(email: str) -> Optional[str]:
registrations = load_registrations()
email_entries = [r for r in registrations if r["email"] == email]
if not email_entries:
return None
if not config["registration"].get("multiple_users_per_email", True):
return "This email address has already been used to register an account."
email_cooldown = config["registration"].get("email_cooldown")
if email_cooldown:
latest = max(datetime.fromisoformat(e["datetime"]) for e in email_entries)
time_since = datetime.utcnow() - latest
if time_since.total_seconds() < email_cooldown:
wait_time = email_cooldown - time_since.total_seconds()
return f"Please wait {int(wait_time)} seconds before requesting another account."
return None
def load_template(template_path: str) -> str:
try:
with open(os.path.join(BASE_DIR, template_path), "r") as f:
return f.read()
except FileNotFoundError:
raise HTTPException(status_code=500, detail=f"Email template not found: {template_path}")
def build_email_message(token: str, requested_username: str, now: datetime, recipient_email: str) -> EmailMessage:
from sw1tch.utilities.time import get_time_until_reset_str
time_until_reset = get_time_until_reset_str(now)
plain_template = load_template(config["email"]["templates"]["registration_token"]["body"])
html_template = load_template(config["email"]["templates"]["registration_token"]["body_html"])
plain_body = plain_template.format(
homeserver=config["homeserver"], registration_token=token,
requested_username=requested_username, utc_time=now.strftime("%H:%M:%S"),
time_until_reset=time_until_reset
)
html_body = html_template.format(
homeserver=config["homeserver"], registration_token=token,
requested_username=requested_username, utc_time=now.strftime("%H:%M:%S"),
time_until_reset=time_until_reset
)
msg = EmailMessage()
msg.set_content(plain_body)
msg.add_alternative(html_body, subtype="html")
msg["Subject"] = config["email"]["templates"]["registration_token"]["subject"].format(homeserver=config["homeserver"])
msg["From"] = config["email"]["smtp"]["from"]
msg["To"] = recipient_email
return msg
def send_email_message(msg: EmailMessage) -> None:
smtp_conf = config["email"]["smtp"]
try:
with smtplib.SMTP(smtp_conf["host"], smtp_conf["port"]) as server:
if smtp_conf.get("use_tls", True):
server.starttls()
server.login(smtp_conf["username"], smtp_conf["password"])
server.send_message(msg)
logger.info(f"Registration email sent successfully to {msg['To']}")
except Exception as ex:
logger.error(f"Failed to send email: {ex}")
raise HTTPException(status_code=500, detail=f"Error sending email: {ex}")

53
sw1tch/utilities/time.py Normal file
View file

@ -0,0 +1,53 @@
from datetime import datetime, timedelta
from typing import Tuple
from sw1tch import config
def get_current_utc() -> datetime:
return datetime.utcnow()
def get_next_reset_time(now: datetime) -> datetime:
reset_h = config["registration"]["token_reset_time_utc"] // 100
reset_m = config["registration"]["token_reset_time_utc"] % 100
candidate = now.replace(hour=reset_h, minute=reset_m, second=0, microsecond=0)
if candidate <= now:
candidate += timedelta(days=1)
return candidate
def get_downtime_start(next_reset: datetime) -> datetime:
return next_reset - timedelta(minutes=config["registration"]["downtime_before_token_reset"])
def format_timedelta(td: timedelta) -> str:
total_minutes = int(td.total_seconds() // 60)
hours = total_minutes // 60
minutes = total_minutes % 60
parts = []
if hours == 1:
parts.append("1 hour")
elif hours > 1:
parts.append(f"{hours} hours")
if minutes == 1:
parts.append("1 minute")
elif minutes > 1:
parts.append(f"{minutes} minutes")
return " and ".join(parts) if parts else "0 minutes"
def get_time_until_reset_str(now: datetime) -> str:
nr = get_next_reset_time(now)
delta = nr - now
return format_timedelta(delta)
def is_registration_closed(now: datetime) -> Tuple[bool, str]:
nr = get_next_reset_time(now)
ds = get_downtime_start(nr)
if ds <= now < nr:
time_until_open = nr - now
msg = f"Registration is closed. It reopens in {format_timedelta(time_until_open)} at {nr.strftime('%H:%M UTC')}."
return True, msg
else:
if now > ds:
nr += timedelta(days=1)
ds = get_downtime_start(nr)
time_until_close = ds - now
msg = f"Registration is open. It will close in {format_timedelta(time_until_close)} at {ds.strftime('%H:%M UTC')}."
return False, msg