Phase 7: merge improvements branch — tests, data-loss guards, LoRa sync, storage optimization

This commit is contained in:
lichenblankie 2026-06-05 04:49:33 +00:00
commit a24fd589b6
34 changed files with 2886 additions and 557 deletions

View file

@ -1,5 +1,15 @@
__pycache__/
**/__pycache__/
*.pyc
index.db*
index.hnsw
tinyweb_identity
.git/
.gitignore
*.md
.env
.env.*
.venv/
venv/
models/
.DS_Store

View file

@ -6,52 +6,76 @@ on:
jobs:
build:
runs-on: docker
runs-on: ubuntu-latest
steps:
- name: Checkout code
uses: actions/checkout@v4
uses: https://code.forgejo.org/actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: '3.11'
- name: Install dependencies
run: |
pip install -r requirements.txt
pip install pyinstaller
apt-get update && apt-get install -y python3 python3-pip python3-venv jq curl
curl -fsSL https://get.docker.com | sh
pip3 install --break-system-packages -r requirements.txt
pip3 install --break-system-packages pyinstaller
- name: Build with PyInstaller
run: |
pyinstaller --onefile --console --name TinyWeb app.py
- name: Upload artifact
uses: actions/upload-artifact@v4
with:
name: TinyWeb-linux-x64
path: dist/TinyWeb
if-no-files-found: error
- name: Prepare artifact
run: |
cp dist/TinyWeb TinyWeb-linux-x64
chmod +x TinyWeb-linux-x64
ls -la TinyWeb-linux-x64
release:
needs: build
runs-on: docker
- name: Get Release ID
if: startsWith(github.ref, 'refs/tags/v')
id: release
run: |
TAG="${{ github.ref_name }}"
REPO="${{ github.repository }}"
TOKEN="${{ secrets.FORGEJO_TOKEN }}"
RELEASE_JSON=$(curl -s "https://git.derickphan.com/api/v1/repos/$REPO/releases/tags/$TAG" \
-H "Authorization: token $TOKEN")
echo "$RELEASE_JSON"
RELEASE_ID=$(echo "$RELEASE_JSON" | jq -r '.id')
echo "release_id=$RELEASE_ID" >> $FORGEJO_OUTPUT
steps:
- name: Download artifact
uses: actions/download-artifact@v4
with:
name: TinyWeb-linux-x64
- name: Upload to Release
if: startsWith(github.ref, 'refs/tags/v')
run: |
FILE=TinyWeb-linux-x64
RELEASE_ID="${{ steps.release.outputs.release_id }}"
REPO="${{ github.repository }}"
TOKEN="${{ secrets.FORGEJO_TOKEN }}"
curl -X POST "https://git.derickphan.com/api/v1/repos/$REPO/releases/$RELEASE_ID/assets" \
-H "Authorization: token $TOKEN" \
-F "attachment=@$FILE"
- name: Make executable
run: chmod +x TinyWeb-linux-x64
- name: Login to Registry
run: |
echo "${{ secrets.REGISTRY_TOKEN }}" | docker login registry.derickphan.com -u _ --password-stdin
- name: Create Release
uses: actions/forgejo-release@v2
with:
direction: upload
release-dir: .
override: true
prerelease: ${{ contains(github.ref_name, 'alpha') || contains(github.ref_name, 'beta') }}
- name: Build and push Docker image
run: |
TAG="${{ github.ref_name }}"
if [ -z "$TAG" ]; then
TAG="latest"
fi
# Configure Docker daemon with DNS
mkdir -p ~/.docker
cat > ~/.docker/daemon.json << 'EOF'
{
"dns": ["8.8.8.8", "1.1.1.1"],
"builder": {
"features": {
"buildkit": true
}
}
}
EOF
# Build with buildkit
DOCKER_BUILDKIT=1 docker build --network=host -t registry.derickphan.com/tinyweb:$TAG .
docker push registry.derickphan.com/tinyweb:$TAG

View file

@ -1,75 +0,0 @@
name: Build
on:
push:
tags:
- 'v*.*.*'
workflow_dispatch:
jobs:
build:
strategy:
fail-fast: false
matrix:
include:
- os: windows-latest
artifact: TinyWeb-windows-x64.exe
- os: macos-latest
artifact: TinyWeb-macos-arm64
- os: ubuntu-latest
artifact: TinyWeb-linux-x64
runs-on: ${{ matrix.os }}
steps:
- name: Checkout code
uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: '3.11'
- name: Install dependencies
run: |
pip install -r requirements.txt
pip install pyinstaller
- name: Build with PyInstaller
run: |
pyinstaller --onefile --console --name TinyWeb app.py
- name: Get artifact path
id: artifact
run: |
if [[ "${{ matrix.os }}" == "windows-latest" ]]; then
echo "path=dist/TinyWeb.exe" >> $GITHUB_OUTPUT
else
echo "path=dist/TinyWeb" >> $GITHUB_OUTPUT
fi
- name: Create ZIP
uses: actions/upload-artifact@v4
with:
name: ${{ matrix.artifact }}
path: ${{ steps.artifact.outputs.path }}
if-no-files-found: error
release:
needs: build
runs-on: ubuntu-latest
if: startsWith(github.ref, 'refs/tags/v')
steps:
- name: Download all artifacts
uses: actions/download-artifact@v4
with:
path: artifacts
- name: Create Release
uses: softprops/action-gh-release@v1
with:
files: artifacts/**
generate_release_notes: true
env:
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}

View file

@ -2,6 +2,12 @@ FROM python:3.12-slim
WORKDIR /app
# Install build tools for packages like hnswlib
RUN apt-get update && apt-get install -y --no-install-recommends \
g++ \
gcc \
&& rm -rf /var/lib/apt/lists/*
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

121
README.md
View file

@ -12,9 +12,33 @@ A personal, decentralized search engine built on the [Reticulum](https://reticul
- **Import/export** — JSON-based backup and restore
- **Mesh-native** — Works over Reticulum without the internet; encrypted and decentralized by default
## Performance & Scale
### Search Speed
| Pages indexed | Search speed | Notes |
|--------------|-------------|-------|
| 1,000 | ~50ms | Fast local FTS5 |
| 10,000 | ~50-100ms | Full-text search |
| 100,000 | ~100-200ms | Combined BM25 + semantic |
| 500,000 | ~200-400ms | With semantic enabled |
| 1,000,000 | ~300-500ms | Hybrid search |
*Times are estimates for combined BM25 + semantic search. Actual performance varies by hardware, storage type (SSD/HDD), and search complexity.*
### Concurrent Connections
- Database pool: 16 simultaneous connections
- Suitable for single-user + a few subscriptions
### Export
- Paginated at 10,000 pages per request
- Use `?batch=N` to export in chunks: `/export?batch=0`, `/export?batch=1`, etc.
## Download (pre-built binaries)
Download the latest release for your platform from the [GitHub Releases](https://github.com/anomalyco/tinyweb/releases) page:
Download the latest release for your platform from the [Releases](https://git.derickphan.com/lichenblankie/tinyweb/releases) page:
| Platform | File |
|----------|------|
@ -24,8 +48,56 @@ Download the latest release for your platform from the [GitHub Releases](https:/
Run the downloaded file — no installation required.
## Docker
Pull and run TinyWeb from the container registry:
```bash
docker run -p 8080:8080 registry.derickphan.com/tinyweb:latest
```
Or with a specific version:
```bash
docker run -p 8080:8080 registry.derickphan.com/tinyweb:v0.1.0
```
### Docker Compose
```yaml
services:
tinyweb:
image: registry.derickphan.com/tinyweb:latest
ports:
- "8080:8080"
volumes:
- tinyweb-data:/data
volumes:
tinyweb-data:
```
Run with `docker compose up -d`.
### Storage Estimates
Average web page content is ~15KB per page:
| Pages | Database | Embeddings* | Total |
|-------|----------|------------|-------|
| 10,000 | 150MB | 80MB | ~250MB |
| 100,000 | 1.5GB | 800MB | ~2.5GB |
| 500,000 | 7.5GB | 4GB | ~12GB |
| 1,000,000 | 15GB | 8GB | ~25GB |
*Embeddings require semantic search to be enabled. With compression enabled (Settings > Search > AI), embeddings use ~50% less storage.
Enable optional compression in Settings > Search > AI to reduce embedding storage by ~50%.
## Data storage
### Local (Python/binary)
Your data is stored in `~/.tinyweb/`:
| File | Description |
@ -37,13 +109,37 @@ Your data is stored in `~/.tinyweb/`:
This allows your data to persist between upgrades and stay separate from the application.
### Backups
Back up the whole `~/.tinyweb/` directory periodically. The two files that matter:
- **`tinyweb_identity`** is your permanent mesh identity. If you lose it, your destination hash changes and every subscriber has to re-subscribe to the new one. Keep it somewhere you trust; the file is `0600` by default.
- **`index.db`** is your full reading history — every page, note, tag, and synced remote page. Losing it loses everything you've curated.
`models/` and `index.hnsw` are re-derivable (the model will re-download, and the HNSW index rebuilds from the database on next startup with semantic search enabled) so they don't need to be backed up.
The `/export` page produces a JSON dump of your pages. It's a migration aid — it doesn't preserve your identity file, your custom template, or subscription state. A full restore needs a copy of `~/.tinyweb/`.
### Docker
Data is stored in the `/data` volume inside the container. Use a volume mount to persist data:
```bash
docker run -p 8080:8080 -v tinyweb-data:/data registry.derickphan.com/tinyweb:latest
```
Or with docker-compose (see above) — data persists in the named volume.
### Command line options
```bash
./TinyWeb --version # Show version
./TinyWeb -p 9000 # Use port 9000 instead of default 8080
./TinyWeb --bind 0.0.0.0 # Expose the web UI to your LAN (see warning below)
```
By default, the web UI binds to `127.0.0.1` and is only reachable from the machine running TinyWeb. **The UI has no authentication** — anyone who can reach the port can read, add, and delete entries, and change settings. Only pass `--bind 0.0.0.0` if you fully trust your network, or put TinyWeb behind an authenticating reverse proxy.
## Getting started
```bash
@ -51,7 +147,7 @@ pip install -r requirements.txt
python app.py
```
This starts the Reticulum server and an HTTP gateway on `http://localhost:8080`. Open it in your browser.
This starts the Reticulum server and an HTTP gateway on `http://127.0.0.1:8080`. Open it in your browser. The UI is localhost-only by default; see `--bind` under *Command line options* if you want to reach it from another machine.
Your destination hash is printed on startup — share it with friends so they can subscribe to your index.
@ -86,7 +182,9 @@ themes/ — Saved HTML templates (e.g. kodama.html)
## Security
TinyWeb includes several hardening measures:
**The web UI has no authentication.** It is bound to `127.0.0.1` by default, so only processes on the local machine can reach it. If you pass `--bind 0.0.0.0` (or run inside a container with a published port), anyone who can reach that address can fully control your instance — reading private entries, changing settings, and modifying the HTML template (which runs in your browser). Put TinyWeb behind a reverse proxy with auth before exposing it beyond localhost.
Other hardening measures:
- **CSRF protection** — All POST forms use per-session tokens via double-submit cookies
- **SSRF prevention** — URL fetching validates hostnames against private IP ranges, with redirect re-validation
@ -96,6 +194,23 @@ TinyWeb includes several hardening measures:
- **Bookmark authentication** — The bookmarklet endpoint requires a secret token
- **Identity file protection** — The Reticulum identity key is restricted to owner-only permissions (0600)
## Maintenance
### Database Vacuum
Over time, deleted pages leave empty space in the database. Run the vacuum tool periodically to reclaim space:
1. Go to `/style` in your browser
2. Click "vacuum database" at the bottom of the page
### Optional Compression
To reduce storage for semantic search embeddings (~50% savings):
1. Go to `/style` > Search > AI
2. Enable "compress embeddings"
3. Re-index your existing pages for the compression to apply to existing embeddings
## Dependencies
- [requests](https://docs.python-requests.org/) — HTTP fetching

132
app.py
View file

@ -8,7 +8,8 @@ from http.server import HTTPServer
from db import init_db, get_setting, set_setting
from handlers import dispatch_request
from gateway import GatewayState, GatewayHandler, GATEWAY_PORT
import gateway
from gateway import GatewayState, GatewayHandler
APP_NAME = "tinyweb"
ASPECTS = ["server"]
@ -24,13 +25,13 @@ def get_transport_config():
return host, int(port)
def find_available_port(start=8080, max_attempts=20):
def find_available_port(start=8080, max_attempts=20, host="127.0.0.1"):
"""Find an available port starting from start."""
import socket
for port in range(start, start + max_attempts):
try:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.bind(("0.0.0.0", port))
s.bind((host, port))
return port
except OSError:
continue
@ -71,30 +72,62 @@ def load_or_create_identity():
return identity
# Remote peers on the Reticulum mesh can only reach a narrow, read-only surface.
# Any other method/path is rejected here — CSRF cannot authenticate mesh callers
# (the attacker controls both the "cookie" and the "form" side of the check), so
# gating by whitelist is the only safe option.
_RNS_ALLOWED = {("GET", "/api/sites")}
def rns_request_handler(path, data, request_id, link_id, remote_identity, requested_at):
if data is None:
data = {"method": "GET", "path": "/", "query": {}, "body": {}, "gateway_host": ""}
method = data.get("method", "GET")
req_path = data.get("path", "/")
if (method, req_path) not in _RNS_ALLOWED:
return {
"status": 403,
"content_type": "text/plain; charset=utf-8",
"body": "Forbidden: this endpoint is not available over Reticulum.",
"headers": {},
}
return dispatch_request(data)
def start_gateway(reticulum):
def start_gateway(reticulum, bind_host="127.0.0.1"):
GatewayState.reticulum = reticulum
GatewayState.local_dispatch = dispatch_request
server = HTTPServer(("0.0.0.0", GATEWAY_PORT), GatewayHandler)
server = HTTPServer((bind_host, gateway.GATEWAY_PORT), GatewayHandler)
thread = threading.Thread(target=server.serve_forever, daemon=True)
thread.start()
def _transport_settings_match(config_file, desired_host, desired_port):
"""Check if existing config transport settings match desired values."""
def _config_settings_match(config_file, desired_host, desired_port):
"""Check if existing config transport and LoRa settings match desired values."""
import configparser
try:
config = configparser.ConfigParser()
config.read(config_file)
if config.has_section("TCP Transport"):
existing_host = config.get("TCP Transport", "target_host")
existing_port = config.get("TCP Transport", "target_port")
return existing_host == desired_host and existing_port == str(desired_port)
# Check TCP transport
tcp_enabled = get_setting("tcp_enabled", "1") == "1"
has_tcp = config.has_section("TCP Transport")
if tcp_enabled != has_tcp:
return False
if tcp_enabled and has_tcp:
if (config.get("TCP Transport", "target_host") != desired_host or
config.get("TCP Transport", "target_port") != str(desired_port)):
return False
# Check LoRa
lora_enabled = get_setting("lora_enabled", "0") == "1"
has_lora = config.has_section("RNode LoRa")
if lora_enabled != has_lora:
return False
if lora_enabled and has_lora:
if config.get("RNode LoRa", "port", fallback="") != get_setting("lora_port", ""):
return False
if config.get("RNode LoRa", "frequency", fallback="") != get_setting("lora_frequency", "867200000"):
return False
return True
except Exception:
pass
return False
@ -110,13 +143,60 @@ def ensure_rns_config(config_dir, transport_host=None, transport_port=None):
if transport_port is None:
transport_port = int(get_setting("transport_port", str(DEFAULT_TRANSPORT_PORT)))
managed_sentinel = "# managed by tinyweb"
if os.path.exists(config_file):
if _transport_settings_match(config_file, transport_host, transport_port):
try:
with open(config_file) as f:
existing = f.read()
except OSError:
existing = ""
if managed_sentinel not in existing:
# User-authored config — don't clobber it.
if not _config_settings_match(config_file, transport_host, transport_port):
print(
f"Warning: {config_file} was not created by tinyweb; "
"leaving it alone. Edit it manually to change transport/LoRa settings."
)
return
if _config_settings_match(config_file, transport_host, transport_port):
return
# Build optional interface blocks
tcp_block = ""
if get_setting("tcp_enabled", "1") == "1":
tcp_block = f"""
[[TCP Transport]]
type = TCPClientInterface
enabled = yes
target_host = {transport_host}
target_port = {transport_port}
"""
lora_block = ""
if get_setting("lora_enabled", "0") == "1":
lora_port = get_setting("lora_port", "")
if lora_port:
lora_frequency = get_setting("lora_frequency", "867200000")
lora_bandwidth = get_setting("lora_bandwidth", "125000")
lora_txpower = get_setting("lora_txpower", "7")
lora_sf = get_setting("lora_sf", "8")
lora_cr = get_setting("lora_cr", "5")
lora_block = f"""
[[RNode LoRa]]
type = RNodeInterface
enabled = yes
port = {lora_port}
frequency = {lora_frequency}
bandwidth = {lora_bandwidth}
txpower = {lora_txpower}
spreadingfactor = {lora_sf}
codingrate = {lora_cr}
"""
os.makedirs(config_dir, exist_ok=True)
with open(config_file, "w") as f:
f.write(f"""[reticulum]
f.write(f"""{managed_sentinel}
[reticulum]
enable_transport = False
share_instance = No
@ -127,13 +207,7 @@ def ensure_rns_config(config_dir, transport_host=None, transport_port=None):
[[Default Interface]]
type = AutoInterface
enabled = Yes
[[TCP Transport]]
type = TCPClientInterface
enabled = yes
target_host = {transport_host}
target_port = {transport_port}
""")
{tcp_block}{lora_block}""")
print(f"Created Reticulum config at {config_file}")
@ -159,15 +233,20 @@ def main():
parser = argparse.ArgumentParser(prog="tinyweb", description="Personal decentralized search engine")
parser.add_argument("--version", "-v", action="store_true", help="Show version")
parser.add_argument("--port", "-p", type=int, default=None, help="HTTP gateway port (default: 8080)")
parser.add_argument(
"--bind", "-b", default="127.0.0.1",
help="Address to bind the HTTP gateway to (default: 127.0.0.1). "
"Use 0.0.0.0 to expose to the LAN; note that the web UI has no authentication.",
)
args = parser.parse_args()
if args.version:
print(f"TinyWeb {get_version()}")
return
bind_host = args.bind
port = args.port or 8080
import gateway
gateway.GATEWAY_PORT = find_available_port(port)
gateway.GATEWAY_PORT = find_available_port(port, host=bind_host)
init_db()
transport_host = get_setting("transport_host", DEFAULT_TRANSPORT_HOST)
@ -196,10 +275,15 @@ def main():
time.sleep(2)
destination.announce()
set_setting("dest_hash", destination.hash.hex())
start_gateway(reticulum)
start_gateway(reticulum, bind_host=bind_host)
print(f"TinyWeb running!")
print(f"Open http://localhost:{GATEWAY_PORT} in your browser")
if bind_host in ("0.0.0.0", "::"):
print(f"Open http://localhost:{gateway.GATEWAY_PORT} in your browser")
print(f"WARNING: listening on {bind_host} — the web UI has no authentication. "
"Anyone on your network can control this instance.")
else:
print(f"Open http://{bind_host}:{gateway.GATEWAY_PORT} in your browser")
print(f"Destination hash: {RNS.prettyhexrep(destination.hash)} (share this so friends can subscribe)")
while True:

128
conftest.py Normal file
View file

@ -0,0 +1,128 @@
"""Shared pytest fixtures for TinyWeb tests.
Three fixtures cover most tests: `temp_db` swaps the SQLite path to a
per-test tempfile, `seeded_db` layers sample rows on top, and `csrf_session`
primes the thread-local CSRF token that handlers read.
"""
import socket
import sys
from pathlib import Path
import pytest
sys.path.insert(0, str(Path(__file__).parent))
import db as db_module
import handlers as handlers_module
@pytest.fixture
def temp_db(tmp_path, monkeypatch):
"""Isolated SQLite DB per test.
Swaps `db.DATABASE` and `db.DATA_DIR` to a tempdir, clears the connection
pool before and after so state doesn't leak across tests, and calls
`init_db()` so every schema object exists.
"""
data_dir = tmp_path / "tinyweb"
data_dir.mkdir()
db_path = data_dir / "index.db"
monkeypatch.setattr(db_module, "DATA_DIR", str(data_dir))
monkeypatch.setattr(db_module, "DATABASE", str(db_path))
with db_module._pool_lock:
for conn in db_module._pool:
try:
conn.close()
except Exception:
pass
db_module._pool.clear()
db_module.init_db()
yield db_path
with db_module._pool_lock:
for conn in db_module._pool:
try:
conn.close()
except Exception:
pass
db_module._pool.clear()
@pytest.fixture
def seeded_db(temp_db):
"""A temp DB with a small, realistic set of pages/tags/links."""
db = db_module.get_db()
try:
rows = [
("https://example.com/rust-intro", "Rust Intro", "A gentle introduction to rust borrow checker.", "notes on ownership"),
("https://example.com/python-tips", "Python Tips", "Daily python tricks for readable code.", ""),
("https://example.com/ocaml-why", "Why OCaml", "Type systems and inference in ocaml.", "private thoughts"),
("https://news.example.org/mesh", "Mesh Networking", "Reticulum and LoRa for decentralized networks.", ""),
]
for url, title, body, note in rows:
db.execute(
"INSERT INTO pages (url, title, body, note, last_modified) "
"VALUES (?, ?, ?, ?, '2026-04-01T00:00:00')",
(url, title, body, note),
)
db.commit()
page_ids = {
row["url"]: row["id"]
for row in db.execute("SELECT id, url FROM pages").fetchall()
}
tag_rows = [
(page_ids["https://example.com/rust-intro"], ["rust", "public"]),
(page_ids["https://example.com/python-tips"], ["python"]),
(page_ids["https://example.com/ocaml-why"], ["ocaml", "private"]),
(page_ids["https://news.example.org/mesh"], ["mesh", "public"]),
]
for pid, tags in tag_rows:
for name in tags:
db.execute("INSERT OR IGNORE INTO tags (name) VALUES (?)", (name,))
tid = db.execute("SELECT id FROM tags WHERE name = ?", (name,)).fetchone()[0]
db.execute(
"INSERT OR IGNORE INTO page_tags (page_id, tag_id) VALUES (?, ?)",
(pid, tid),
)
db.execute(
"INSERT INTO links (page_id, url, label) VALUES (?, ?, ?)",
(page_ids["https://example.com/rust-intro"], "https://example.com/rust-advanced", "advanced rust guide"),
)
db.commit()
finally:
db_module.return_db(db)
return temp_db
@pytest.fixture
def csrf_session(monkeypatch):
"""Prime the CSRF thread-local so handler code that calls _get_csrf_token works."""
token = "test-csrf-token"
handlers_module._request_local.csrf_token = token
yield token
if hasattr(handlers_module._request_local, "csrf_token"):
del handlers_module._request_local.csrf_token
def patch_dns_fail(monkeypatch):
"""Make every socket.getaddrinfo call raise gaierror for the rest of this test."""
def boom(*args, **kwargs):
raise socket.gaierror("test: DNS disabled")
monkeypatch.setattr(socket, "getaddrinfo", boom)
def patch_dns_ok(monkeypatch, address="93.184.216.34"):
"""Make every getaddrinfo return a single public IP for the rest of this test."""
def ok(host, port, *args, **kwargs):
return [(socket.AF_INET, socket.SOCK_STREAM, 0, "", (address, port or 80))]
monkeypatch.setattr(socket, "getaddrinfo", ok)
def patch_dns_private(monkeypatch, address="127.0.0.1"):
"""Make every getaddrinfo return a private/blocked IP for the rest of this test."""
def private(host, port, *args, **kwargs):
return [(socket.AF_INET, socket.SOCK_STREAM, 0, "", (address, port or 80))]
monkeypatch.setattr(socket, "getaddrinfo", private)

37
db.py
View file

@ -70,10 +70,16 @@ def clean_url(url):
# Prefer https
scheme = "https" if parsed.scheme in ("http", "https") else parsed.scheme
# Normalize hostname: lowercase, strip www.
# Normalize hostname: lowercase, strip www (only if non-www resolves)
hostname = (parsed.hostname or "").lower()
original_hostname = hostname
if hostname.startswith("www."):
hostname = hostname[4:]
port = parsed.port or (443 if scheme == "https" else 80)
try:
socket.getaddrinfo(hostname, port, proto=socket.IPPROTO_TCP)
except socket.gaierror:
hostname = original_hostname
# Preserve explicit non-default ports
port = parsed.port
@ -97,7 +103,7 @@ def clean_url(url):
_pool = []
_pool_lock = __import__("threading").Lock()
_POOL_SIZE = 4
_POOL_SIZE = 16
def get_db():
@ -117,6 +123,14 @@ def get_db():
def return_db(db):
try:
db.rollback()
except Exception:
try:
db.close()
except Exception:
pass
return
with _pool_lock:
if len(_pool) < _POOL_SIZE:
_pool.append(db)
@ -271,8 +285,15 @@ def init_db():
)
db.execute("CREATE INDEX IF NOT EXISTS idx_chunks_page ON chunks(page_id)")
db.execute("CREATE INDEX IF NOT EXISTS idx_chunks_remote ON chunks(remote_page_id)")
db.execute("CREATE INDEX IF NOT EXISTS idx_chunks_page_idx ON chunks(page_id, chunk_index)")
db.execute("CREATE INDEX IF NOT EXISTS idx_pages_url ON pages(url)")
db.execute("CREATE INDEX IF NOT EXISTS idx_pages_modified ON pages(last_modified)")
db.execute("CREATE INDEX IF NOT EXISTS idx_page_tags_page ON page_tags(page_id)")
db.execute("CREATE INDEX IF NOT EXISTS idx_page_tags_tag ON page_tags(tag_id)")
db.execute("PRAGMA journal_mode=WAL")
db.execute("PRAGMA synchronous=NORMAL")
db.execute("PRAGMA cache_size=-64000")
db.commit()
db.close()
@ -286,6 +307,16 @@ def get_setting(key, default=""):
return_db(db)
def vacuum_db():
"""Run VACUUM and WAL checkpoint to reclaim space after deletions."""
db = get_db()
try:
db.execute("PRAGMA wal_checkpoint(TRUNCATE)")
db.execute("VACUUM")
finally:
return_db(db)
def set_setting(key, value):
db = get_db()
try:
@ -389,7 +420,7 @@ def index_url(url, note="", reticulum_dest=""):
(page_id, href, label),
)
db.commit()
if get_setting("semantic_search", "1") == "1":
if get_setting("semantic_search", "0") == "1":
try:
from embeddings import store_embeddings
store_embeddings(page_id, title, body, db)

View file

@ -233,24 +233,49 @@ def embed(texts, is_query=False):
"token_type_ids": token_type_ids,
},
)
# CLS token pooling — take the first token's hidden state
emb = outputs[0][:, 0, :]
all_embeddings.append(emb)
embeddings = np.concatenate(all_embeddings, axis=0)
# L2 normalize
norms = np.linalg.norm(embeddings, axis=1, keepdims=True)
norms = np.maximum(norms, 1e-12)
embeddings = embeddings / norms
return _maybe_compress(embeddings.astype(np.float32))
def _maybe_compress(embeddings):
"""Compress embeddings to float16 if compression is enabled."""
try:
from db import get_setting
if get_setting("compress_embeddings", "0") == "1":
return embeddings.astype(np.float16)
except Exception:
pass
return embeddings
def _decompress(embeddings):
"""Decompress float16 embeddings to float32 if needed."""
if embeddings.dtype == np.float16:
return embeddings.astype(np.float32)
return embeddings
def _blob_to_vec(buf):
"""Decode a stored embedding blob to a float32 vector, inferring dtype from length."""
if len(buf) == DIMS * 2:
return np.frombuffer(buf, dtype=np.float16).astype(np.float32)
return np.frombuffer(buf, dtype=np.float32)
# ---------------------------------------------------------------------------
# HNSW index management
# ---------------------------------------------------------------------------
BATCH_SIZE = 50000
def build_index(db=None):
"""Load all embeddings from chunks table and build HNSW index."""
"""Load all embeddings from chunks table and build HNSW index in batches."""
import hnswlib
global _hnsw_index, _hnsw_ids
@ -258,29 +283,47 @@ def build_index(db=None):
own_db = db is None
if own_db:
db = get_db()
try:
rows = db.execute("SELECT id, embedding FROM chunks ORDER BY id").fetchall()
finally:
if own_db:
return_db(db)
try:
total = db.execute("SELECT COUNT(*) FROM chunks").fetchone()[0]
if total == 0:
with _hnsw_lock:
if not rows:
_hnsw_index = None
_hnsw_ids = []
return
n = len(rows)
ids = [r["id"] for r in rows]
matrix = np.frombuffer(b"".join(r["embedding"] for r in rows), dtype=np.float32).reshape(n, DIMS)
all_ids = []
all_embeddings = []
for offset in range(0, total, BATCH_SIZE):
rows = db.execute(
"SELECT id, embedding FROM chunks ORDER BY id LIMIT ? OFFSET ?",
(BATCH_SIZE, offset),
).fetchall()
for r in rows:
emb = _blob_to_vec(r["embedding"])
all_ids.append(r["id"])
all_embeddings.append(emb)
finally:
if own_db:
return_db(db)
if not all_ids:
with _hnsw_lock:
_hnsw_index = None
_hnsw_ids = []
return
matrix = np.stack(all_embeddings)
n = len(all_ids)
ids = all_ids
index = hnswlib.Index(space="cosine", dim=DIMS)
# ef_construction and M balance build speed vs recall;
# these defaults give >99% recall at reasonable build time
index.init_index(max_elements=max(n, 1024), ef_construction=200, M=16)
index.add_items(matrix, list(range(n)))
index.set_ef(50) # query-time accuracy parameter
index.set_ef(50)
with _hnsw_lock:
_hnsw_index = index
_hnsw_ids = ids
@ -319,8 +362,8 @@ def store_embeddings(page_id, title, body, db):
return
embeddings_matrix = embed(chunks)
embeddings_matrix = _decompress(embeddings_matrix)
# Delete old chunks for this page
db.execute("DELETE FROM chunks WHERE page_id = ?", (page_id,))
new_ids = []
@ -343,6 +386,7 @@ def store_remote_embeddings(remote_page_id, title, note, db):
return
embeddings_matrix = embed([text])
embeddings_matrix = _decompress(embeddings_matrix)
db.execute("DELETE FROM chunks WHERE remote_page_id = ?", (remote_page_id,))
cursor = db.execute(

View file

@ -30,4 +30,5 @@ EOF
fi
export RNS_CONFIG_DIR="$CONFIG_DIR"
exec python app.py
# Bind to 0.0.0.0 inside the container; isolation is handled by Docker's port mapping.
exec python app.py --bind 0.0.0.0 "$@"

View file

@ -1,3 +1,4 @@
import re
import sys
import time
import threading
@ -9,6 +10,7 @@ APP_NAME = "tinyweb"
ASPECTS = ["server"]
GATEWAY_PORT = 8080
REQUEST_TIMEOUT = 60
MAX_BODY_SIZE = 16 * 1024 * 1024 # 16 MiB — covers /import and every other form
class GatewayState:
@ -71,8 +73,18 @@ class GatewayHandler(BaseHTTPRequestHandler):
body = {}
if method == "POST":
try:
length = int(self.headers.get("Content-Length", 0))
raw = self.rfile.read(length).decode()
except ValueError:
self.send_error(400, "Invalid Content-Length")
return
if length < 0:
self.send_error(400, "Invalid Content-Length")
return
if length > MAX_BODY_SIZE:
self.send_error(413, "Request body too large")
return
raw = self.rfile.read(length).decode("utf-8", errors="replace")
body = parse_qs(raw)
# Parse cookies
@ -123,6 +135,14 @@ class GatewayHandler(BaseHTTPRequestHandler):
self.send_response(resp["status"])
self.send_header("Content-Type", resp.get("content_type", "text/html; charset=utf-8"))
self.send_header("Referrer-Policy", "no-referrer")
self.send_header("X-Content-Type-Options", "nosniff")
self.send_header("X-Frame-Options", "DENY")
self.send_header("Content-Security-Policy",
"default-src 'self'; "
"style-src 'self' 'unsafe-inline'; "
"script-src 'self' 'unsafe-inline'; "
"img-src 'self' data:")
for k, v in resp.get("headers", {}).items():
self.send_header(k, v)
self.end_headers()
@ -144,7 +164,14 @@ class GatewayHandler(BaseHTTPRequestHandler):
self._forward("POST")
def log_message(self, format, *args):
print(f"[Gateway] {args[0]}")
try:
msg = format % args
except TypeError:
msg = format
# /bookmark carries a long-lived token and the URL being indexed —
# redact the query so it doesn't end up in stdout, journald, docker logs, etc.
msg = re.sub(r'(/bookmark)\?\S*', r'\1?[redacted]', msg)
print(f"[Gateway] {msg}")
def main():

View file

@ -50,14 +50,18 @@ def _sanitize_fts_query(query):
if not words:
return '""'
tokens = []
last_idx = len(words) - 1
for i, w in enumerate(words):
# Strip FTS5 special characters to prevent injection
cleaned = re.sub(r'["\'\(\)\*\+\-\^~]', '', w).strip()
# Strip FTS5 special characters (operators, column filter colon) to prevent injection
cleaned = re.sub(r'["\'\(\)\*\+\-\^~:]', '', w).strip()
if not cleaned:
continue
if cleaned.lower() in _STOPWORDS:
continue
if i == len(words) - 1:
# Drop FTS5 operator words so they aren't parsed as operators on the unquoted last token
if cleaned.upper() in ("AND", "OR", "NOT", "NEAR"):
continue
if i == last_idx:
# Prefix match on the last token for partial word matching
tokens.append(f"{cleaned}*")
else:
@ -174,6 +178,11 @@ def _set_page_tags(page_id, tag_string, db=None):
return_db(db)
def _cleanup_orphaned_tags(db):
"""Delete tags that have no page associations."""
db.execute("DELETE FROM tags WHERE id NOT IN (SELECT DISTINCT tag_id FROM page_tags)")
# --- Route handlers ---
@ -205,7 +214,7 @@ def handle_search(query):
# Hybrid search: merge BM25 + semantic via RRF
bm25_ids = [r["id"] for r in bm25_rows]
chunk_snippets = {} # page_id -> best chunk text
if get_setting("semantic_search", "1") == "1":
if get_setting("semantic_search", "0") == "1":
try:
from embeddings import hybrid_search
use_reranker = get_setting("use_reranker", "1") == "1"
@ -245,7 +254,7 @@ def handle_search(query):
snip_html = f'<br>{esc(r["summary"])}' if r["summary"] else ""
result_html += (
f'<div class="result">'
f'<a href="{esc(r["url"])}">{esc(r["title"])}</a><br>'
f'<a href="{esc(r["url"])}" rel="noreferrer noopener">{esc(r["title"])}</a><br>'
f'<small>{esc(r["url"])}</small>'
f'{snip_html}'
f'{note_html}{tags_html}'
@ -276,7 +285,7 @@ def handle_search(query):
items = ""
for l in trusted:
items += (
f'<li><a href="{esc(l["url"])}">{esc(l["label"])}</a> '
f'<li><a href="{esc(clean_url(l["url"]))}" rel="noreferrer noopener">{esc(l["label"])}</a> '
f'<small>— from {esc(l["source_title"])}</small></li>'
)
trusted_html = (
@ -311,8 +320,8 @@ def handle_search(query):
for r in items:
note_html = f' — <em>{esc(r["note"])}</em>' if r["note"] else ""
source_items += (
f'<li><a href="{esc(r["url"])}">{esc(r["title"])}</a>'
f'{note_html} <small>({esc(r["url"])})</small></li>'
f'<li><a href="{esc(clean_url(r["url"]))}" rel="noreferrer noopener">{esc(r["title"])}</a>'
f'{note_html} <small>({esc(clean_url(r["url"]))})</small></li>'
)
remote_html += (
f'<details class="remote" open>'
@ -325,6 +334,18 @@ def handle_search(query):
sub_count = ""
if q and remote_rows:
sub_count = f" + {len(remote_rows)} from subscriptions"
welcome_html = ""
if count == 0 and not q:
welcome_html = (
'<section style="margin-top:1.5rem;max-width:40em">'
'<p>Your index is empty.</p>'
'<p>tinyweb is a personal search engine for pages you save. '
'The index stays on your machine; so does every search.</p>'
'<p>From here: <a href="/add">add a page</a>, '
'<a href="/style">get the bookmarklet</a>, or '
'<a href="/subscriptions">subscribe to another instance</a>.</p>'
'</section>'
)
return _respond(
f'<form method="get" action="/">'
f'<input name="q" value="{esc(q)}" placeholder="search your index" size="40">'
@ -332,6 +353,7 @@ def handle_search(query):
f'</form>'
f'<p class="meta">{count} pages indexed'
f' · <a href="/add">+ add url</a></p>'
f'{welcome_html}'
f'{result_html}'
f'{_page_nav(page, total_results, f"/?q={esc(q)}") if q else ""}'
f'{trusted_html}{remote_html}'
@ -359,7 +381,8 @@ def handle_add_form(msg="", action_type="index"):
f'{_csrf_field()}'
f'<input name="url" placeholder="https://example.com" size="50"><br><br>'
f'<input name="note" placeholder="why are you saving this? (optional)" size="50"><br><br>'
f'<input name="tags" placeholder="tags (comma-separated, e.g. solarpunk, mesh)" size="50"><br><br>'
f'<input name="tags" placeholder="tags (comma-separated, e.g. solarpunk, mesh)" size="50"><br>'
f'<small>tag: private to exclude from sharing</small><br><br>'
f'<button type="submit">index</button>'
f"</form>"
f"<p>{msg}</p>"
@ -421,7 +444,7 @@ def handle_add_submit(body):
f'<label>Title:</label><br>'
f'<input name="manual_title" size="50" placeholder="page title" required><br><br>'
f'<label>Description:</label><br>'
f'<textarea name="manual_description" rows="4" cols="50" placeholder="what is this site about?" required></textarea><br><br>'
f'<textarea name="manual_description" rows="4" cols="50" placeholder="what is this site about? (optional)"></textarea><br><br>'
f'<button type="submit">save manually</button>'
f"</form>"
f'<a href="/">back</a>'
@ -439,8 +462,8 @@ def handle_add_manual_submit(body):
if not url:
return handle_add_form("URL is required.")
if not manual_title or not manual_desc:
return handle_add_form("Title and description are required for manual entry.")
if not manual_title:
return handle_add_form("Title is required for manual entry.")
db = get_db()
try:
@ -463,7 +486,7 @@ def handle_add_manual_submit(body):
db.commit()
# Generate embeddings for this page (if semantic search is enabled)
if get_setting("semantic_search", "1") == "1":
if get_setting("semantic_search", "0") == "1":
try:
from embeddings import store_embeddings
# Pass the page_id, title, description, and db connection
@ -473,7 +496,7 @@ def handle_add_manual_submit(body):
# Log error but don't fail the whole operation
print(f"Error generating embeddings: {e}")
return handle_add_form(f'Added manually: <a href="{esc(url)}">{esc(manual_title)}</a>')
return handle_add_form(f'Added manually: <a href="{esc(url)}" rel="noreferrer noopener">{esc(manual_title)}</a>')
finally:
return_db(db)
@ -499,8 +522,9 @@ def handle_pages(query=None):
tag_links = " ".join(f'<a href="/tags/{esc(t)}">[{esc(t)}]</a>' for t in tags)
tags_html = f' {tag_links}'
items += (
f'<li>{esc(r["title"])}{note_html}{tags_html} '
f'<small>(<a href="{esc(r["url"])}">{esc(r["url"])}</a>)</small> '
f'<li><label><input type="checkbox" name="ids" value="{r["id"]}"> '
f'{esc(r["title"])}</label>{note_html}{tags_html} '
f'<small>(<a href="{esc(r["url"])}" rel="noreferrer noopener">{esc(r["url"])}</a>)</small> '
f'<a href="/edit/{r["id"]}">edit</a> '
f'<a href="/delete/{r["id"]}">remove</a></li>'
)
@ -509,13 +533,112 @@ def handle_pages(query=None):
return _respond(
f"<h1>indexed pages ({total})</h1>"
f"{msg_html}"
f'<form method="post" action="/pages/bulk">'
f'{_csrf_field()}'
f'<p><label><input type="checkbox" id="select-all"> select all</label></p>'
f"<ul>{items}</ul>"
f'{_page_nav(page, total, "/pages", BROWSE_PER_PAGE)}'
f'<details><summary>bulk actions</summary>'
f'<p><button type="submit" name="action" value="delete" id="bulk-delete">delete selected</button></p>'
f'<p><input name="bulk_tags" placeholder="tags (comma-separated)" size="40"> '
f'<select name="tag_mode"><option value="add">add tags</option><option value="replace">replace tags</option></select> '
f'<button type="submit" name="action" value="retag">retag selected</button></p>'
f'</details>'
f'</form>'
f'<script>'
f'document.getElementById("select-all").addEventListener("change",function(){{'
f'document.querySelectorAll("input[name=ids]").forEach(function(c){{c.checked=this.checked}}.bind(this))'
f'}});'
f'document.getElementById("bulk-delete").addEventListener("click",function(e){{'
f'var n=document.querySelectorAll("input[name=ids]:checked").length;'
f'if(!n){{e.preventDefault();return}}'
f'if(!confirm("Delete "+n+" selected page"+(n===1?"":"s")+"?"))e.preventDefault()'
f'}});'
f'</script>'
f'<p><a href="/export">export</a> | <a href="/import">import</a></p>'
f'<a href="/">back</a>'
)
def _render_bulk_delete_confirm(page_ids):
"""Server-side confirmation page for bulk deletion — mirrors handle_delete_confirm."""
db = get_db()
try:
placeholders = ",".join("?" * len(page_ids))
rows = db.execute(
f"SELECT id, url, title FROM pages WHERE id IN ({placeholders})",
page_ids,
).fetchall()
finally:
return_db(db)
if not rows:
return _redirect("/pages")
items = "".join(
f'<li><b>{esc(r["title"] or r["url"])}</b><br>'
f'<small>{esc(r["url"])}</small></li>'
for r in rows
)
hidden_ids = "".join(
f'<input type="hidden" name="ids" value="{int(r["id"])}">' for r in rows
)
n = len(rows)
return _respond(
f"<h1>confirm delete</h1>"
f"<p>Remove the following {n} page{'' if n == 1 else 's'}?</p>"
f"<ul>{items}</ul>"
f'<form method="post" action="/pages/bulk">'
f'{_csrf_field()}'
f'{hidden_ids}'
f'<input type="hidden" name="action" value="delete">'
f'<input type="hidden" name="confirmed" value="1">'
f'<button type="submit">yes, delete {n} page{"" if n == 1 else "s"}</button>'
f"</form>"
f' <a href="/pages">cancel</a>'
)
def handle_bulk_action(body):
ids = body.get("ids", [])
action = body.get("action", [""])[0]
if not ids:
return _redirect("/pages")
# Validate all ids are integers
try:
page_ids = [int(i) for i in ids]
except ValueError:
return _error(400)
# Require an explicit second-step confirmation for bulk delete — the JS
# confirm() on /pages is a first-line filter only.
if action == "delete" and body.get("confirmed", [""])[0] != "1":
return _render_bulk_delete_confirm(page_ids)
db = get_db()
try:
if action == "delete":
for pid in page_ids:
db.execute("DELETE FROM page_tags WHERE page_id = ?", (pid,))
db.execute("DELETE FROM links WHERE page_id = ?", (pid,))
db.execute("DELETE FROM pages WHERE id = ?", (pid,))
_cleanup_orphaned_tags(db)
db.commit()
elif action == "retag":
bulk_tags = body.get("bulk_tags", [""])[0].strip()
tag_mode = body.get("tag_mode", ["add"])[0]
if bulk_tags:
for pid in page_ids:
if tag_mode == "add":
existing = _get_page_tags(pid, db)
new_tags = [t.strip().lower() for t in bulk_tags.split(",") if t.strip()]
merged = ", ".join(sorted(set(existing + new_tags)))
_set_page_tags(pid, merged, db)
else:
_set_page_tags(pid, bulk_tags, db)
_cleanup_orphaned_tags(db)
db.commit()
finally:
return_db(db)
return _redirect("/pages")
def handle_edit_form(page_id, msg=""):
db = get_db()
try:
@ -539,7 +662,8 @@ def handle_edit_form(page_id, msg=""):
f'<label>Note (why you saved this):</label><br>'
f'<input name="note" value="{esc(row["note"])}" size="50"><br><br>'
f'<label>Tags (comma-separated):</label><br>'
f'<input name="tags" value="{esc(tags)}" size="50"><br><br>'
f'<input name="tags" value="{esc(tags)}" size="50"> '
f'<small>(tag: private to keep private)</small><br><br>'
f'<button type="submit">save</button>'
f"</form>"
f"<p>{msg}</p>"
@ -561,6 +685,7 @@ def handle_edit_submit(page_id, body):
)
_set_page_tags(page_id, tags, db)
_cleanup_orphaned_tags(db)
db.commit()
@ -596,6 +721,7 @@ def handle_delete(page_id):
db.execute("DELETE FROM page_tags WHERE page_id = ?", (page_id,))
db.execute("DELETE FROM links WHERE page_id = ?", (page_id,))
db.execute("DELETE FROM pages WHERE id = ?", (page_id,))
_cleanup_orphaned_tags(db)
db.commit()
finally:
return_db(db)
@ -618,10 +744,19 @@ def handle_bookmark(query):
return _text_response(msg, headers={"Access-Control-Allow-Origin": "*"})
def handle_export():
MAX_EXPORT = 10000
def handle_export(query=None):
try:
batch = int((query or {}).get("batch", ["0"])[0])
except (TypeError, ValueError):
batch = 0
db = get_db()
try:
rows = db.execute("SELECT url, title, note FROM pages ORDER BY id").fetchall()
rows = db.execute(
"SELECT url, title, note FROM pages ORDER BY id LIMIT ? OFFSET ?",
(MAX_EXPORT, batch * MAX_EXPORT),
).fetchall()
finally:
return_db(db)
data = [{"url": r["url"], "title": r["title"], "note": r["note"]} for r in rows]
@ -678,14 +813,33 @@ def handle_style_form(msg=""):
name = get_site_name()
sharing = get_setting("sharing_enabled", "0")
checked = " checked" if sharing == "1" else ""
sharing_mode = get_setting("sharing_mode", "exclude_private")
exclude_checked = " checked" if sharing_mode != "require_public" else ""
require_checked = " checked" if sharing_mode == "require_public" else ""
shared_count = _count_shared_pages()
semantic = get_setting("semantic_search", "0")
semantic_checked = " checked" if semantic == "1" else ""
reranker = get_setting("use_reranker", "0")
reranker_checked = " checked" if reranker == "1" else ""
disabled = "" if semantic == "1" else " disabled"
dimmed = ' style="opacity:0.4"' if semantic != "1" else ""
tcp_enabled = get_setting("tcp_enabled", "1")
tcp_checked = " checked" if tcp_enabled == "1" else ""
tcp_disabled = "" if tcp_enabled == "1" else " disabled"
transport_host = get_setting("transport_host", "reticulum.derickphan.com")
transport_port = get_setting("transport_port", "4242")
compress = get_setting("compress_embeddings", "0")
compress_checked = " checked" if compress == "1" else ""
lora_enabled = get_setting("lora_enabled", "0")
lora_checked = " checked" if lora_enabled == "1" else ""
lora_disabled = "" if lora_enabled == "1" else " disabled"
lora_dimmed = ' style="opacity:0.4"' if lora_enabled != "1" else ""
lora_port = get_setting("lora_port", "")
lora_frequency = get_setting("lora_frequency", "867200000")
lora_bandwidth = get_setting("lora_bandwidth", "125000")
lora_txpower = get_setting("lora_txpower", "7")
lora_sf = get_setting("lora_sf", "8")
lora_cr = get_setting("lora_cr", "5")
return _respond(
f"<h1>customize</h1>"
f"<h2>name your search engine</h2>"
@ -694,13 +848,57 @@ def handle_style_form(msg=""):
f'<input name="site_name" value="{esc(name)}" placeholder="tinyweb" size="30"><br><br>'
f"<h2>sharing</h2>"
f'<label><input type="checkbox" name="sharing_enabled" value="1"{checked}>'
f" share your site list publicly at /api/sites</label><br><br>"
f" share your site list publicly at /api/sites</label><br>"
f'<div style="margin-top:0.6rem">'
f"<small>What to share:</small><br>"
f'<label><input type="radio" name="sharing_mode" value="exclude_private"{exclude_checked}>'
f' share all pages except those tagged <code>private</code></label><br>'
f'<label><input type="radio" name="sharing_mode" value="require_public"{require_checked}>'
f' share only pages tagged <code>public</code></label><br>'
f'<small>The <code>private</code> tag always excludes a page, even in public-only mode.</small>'
f'</div>'
f'<p style="margin-top:0.6rem">'
f'Currently sharing <b>{shared_count}</b> page(s). '
f'<a href="/share/preview">preview what subscribers would see</a>'
f'</p>'
f"<h2>mesh network</h2>"
f"<p>Connect to a Reticulum transport node to reach other peers.</p>"
f"<p>Choose how to connect to the mesh. You can enable both for maximum reach.</p>"
f"<h3>internet</h3>"
f'<label><input type="checkbox" name="tcp_enabled" value="1"{tcp_checked} '
f'onchange="var d=!this.checked;'
f'for(var e of document.querySelectorAll(\'#tcp-fields input\'))e.disabled=d;'
f'document.getElementById(\'tcp-fields\').style.opacity=d?\'0.4\':\'1\'">'
f" connect via internet transport node</label><br>"
f"<small>Reach peers anywhere online.</small><br>"
f'<div id="tcp-fields" style="margin-top:0.5rem{";opacity:0.4" if tcp_enabled != "1" else ""}">'
f"<small>Default: reticulum.derickphan.com:4242</small><br>"
f'<input name="transport_host" value="{esc(transport_host)}" placeholder="hostname" size="30">'
f' <input name="transport_port" value="{esc(transport_port)}" placeholder="port" size="6"><br>'
f'<p><a href="https://rmap.world/" target="_blank">discover more nodes</a></p><br>'
f'<input name="transport_host" value="{esc(transport_host)}" placeholder="hostname" size="30"{tcp_disabled}>'
f' <input name="transport_port" value="{esc(transport_port)}" placeholder="port" size="6"{tcp_disabled}><br>'
f'<p><a href="https://rmap.world/" target="_blank" rel="noreferrer noopener">discover more nodes</a></p>'
f'</div><br>'
f"<h3>LoRa</h3>"
f'<label><input type="checkbox" name="lora_enabled" value="1"{lora_checked} '
f'onchange="var d=!this.checked;document.getElementById(\'lora-port\').disabled=d;'
f'document.getElementById(\'lora-extras\').style.opacity=d?\'0.4\':\'1\';'
f'for(var e of document.querySelectorAll(\'#lora-extras input\'))e.disabled=d">'
f" connect via LoRa radio</label><br>"
f"<small>Reach nearby peers off-grid with an <a href=\"https://unsigned.io/rnode/\" target=\"_blank\" rel=\"noreferrer noopener\">RNode</a>.</small><br><br>"
f'<div id="lora-fields" style="{";opacity:0.4" if lora_enabled != "1" else ""}">'
f'<label>Serial port: <input id="lora-port" name="lora_port" value="{esc(lora_port)}" '
f'placeholder="/dev/ttyUSB0" size="20"{lora_disabled}></label><br><br>'
f'<details><summary>advanced radio settings</summary>'
f'<div id="lora-extras" style="margin-top:0.5rem">'
f'<label>Frequency (Hz): <input name="lora_frequency" value="{esc(lora_frequency)}" size="12"{lora_disabled}></label><br>'
f"<small>ISM band frequency. Default: 867200000 (868 MHz EU). US: 915000000.</small><br><br>"
f'<label>Bandwidth (Hz): <input name="lora_bandwidth" value="{esc(lora_bandwidth)}" size="8"{lora_disabled}></label><br>'
f"<small>Default: 125000</small><br><br>"
f'<label>TX Power (dBm): <input name="lora_txpower" value="{esc(lora_txpower)}" size="4"{lora_disabled}></label><br>'
f"<small>0-17 typical. Check local regulations.</small><br><br>"
f'<label>Spreading Factor: <input name="lora_sf" value="{esc(lora_sf)}" size="4"{lora_disabled}></label><br>'
f"<small>5-12. Higher = longer range, slower speed.</small><br><br>"
f'<label>Coding Rate: <input name="lora_cr" value="{esc(lora_cr)}" size="4"{lora_disabled}></label><br>'
f"<small>5-8. Higher = more error correction.</small><br>"
f'</div></details></div><br>'
f"<h2>search</h2>"
f"<h3>ai</h3>"
f'<label><input type="checkbox" name="semantic_search" value="1"{semantic_checked} '
@ -712,6 +910,9 @@ def handle_style_form(msg=""):
f'<label><input type="checkbox" id="reranker" name="use_reranker" value="1"{reranker_checked}{disabled}>'
f" cross-encoder reranking (more accurate)</label><br>"
f"<small>Uses a 22MB model. Adds ~50ms per search. Disable for faster results.</small><br><br>"
f'<label><input type="checkbox" name="compress_embeddings" value="1"{compress_checked}{disabled}>'
f" compress embeddings (50% storage savings)</label><br>"
f"<small>Saves ~50% on storage for embeddings. Slight quality reduction at large scale.</small><br><br>"
f'<a href="/reindex">manage semantic index</a><br><br>'
f"</div>"
f"<h2>custom html</h2>"
@ -724,10 +925,16 @@ def handle_style_form(msg=""):
f"<p>Drag this link to your bookmarks bar. Click it on any page to index it instantly.</p>"
f'<p><a href="javascript:void(fetch(\'http://localhost:8080/bookmark?url=\'+encodeURIComponent(location.href)+\'&token={_get_bookmark_token()}\').then(r=>r.text()).then(t=>alert(t)).catch(()=>alert(\'tinyweb not running\')))">+ save to {esc(name)}</a></p>'
f"<h2>reset</h2>"
f'<form method="post" action="/style/reset">'
f'<form method="post" action="/style/reset" '
f'onsubmit="return confirm(\'Reset the template to default? Your custom template will be lost.\')">'
f'{_csrf_field()}'
f'<button type="submit">reset template to default</button>'
f"</form>"
f"<h2>maintenance</h2>"
f'<form method="post" action="/style/vacuum">'
f'{_csrf_field()}'
f'<button type="submit">vacuum database</button>'
f"</form>"
f"<p>{msg}</p>"
f'<a href="/">back</a>',
use_default=True,
@ -738,8 +945,14 @@ def handle_style_submit(body):
template = body.get("template", [""])[0].replace("\r\n", "\n").replace("\r", "\n")
name = body.get("site_name", ["tinyweb"])[0].strip()
sharing = "1" if body.get("sharing_enabled") else "0"
sharing_mode = body.get("sharing_mode", ["exclude_private"])[0]
if sharing_mode not in ("exclude_private", "require_public"):
sharing_mode = "exclude_private"
set_setting("sharing_mode", sharing_mode)
semantic = "1" if body.get("semantic_search") else "0"
reranker = "1" if body.get("use_reranker") else "0"
compress = "1" if body.get("compress_embeddings") else "0"
tcp_enabled = "1" if body.get("tcp_enabled") else "0"
transport_host = body.get("transport_host", [""])[0].strip()
transport_port = body.get("transport_port", [""])[0].strip()
set_setting("custom_template", template if template.strip() != DEFAULT_TEMPLATE.strip() else "")
@ -747,10 +960,20 @@ def handle_style_submit(body):
set_setting("sharing_enabled", sharing)
set_setting("semantic_search", semantic)
set_setting("use_reranker", reranker)
set_setting("compress_embeddings", compress)
set_setting("tcp_enabled", tcp_enabled)
if transport_host:
set_setting("transport_host", transport_host)
if transport_port:
set_setting("transport_port", transport_port)
lora_enabled = "1" if body.get("lora_enabled") else "0"
set_setting("lora_enabled", lora_enabled)
set_setting("lora_port", body.get("lora_port", [""])[0].strip())
set_setting("lora_frequency", body.get("lora_frequency", ["867200000"])[0].strip())
set_setting("lora_bandwidth", body.get("lora_bandwidth", ["125000"])[0].strip())
set_setting("lora_txpower", body.get("lora_txpower", ["7"])[0].strip())
set_setting("lora_sf", body.get("lora_sf", ["8"])[0].strip())
set_setting("lora_cr", body.get("lora_cr", ["5"])[0].strip())
return handle_style_form("Saved. Restart TinyWeb for mesh network changes to take effect.")
@ -792,6 +1015,22 @@ def handle_about():
f'</ul>'
f'{sharing_html}'
f'{hash_html}'
f'<h2>your data</h2>'
f'<p>Everything is stored locally under <code>~/.tinyweb/</code>:</p>'
f'<ul>'
f'<li><code>tinyweb_identity</code> — your permanent mesh identity. '
f'If you lose this file, your destination hash changes and subscribers '
f'have to re-subscribe to the new one.</li>'
f'<li><code>index.db</code> — your full reading history: every page, '
f'note, tag, and synced remote page.</li>'
f'<li><code>models/</code> — the semantic search model if you enabled it '
f'(redownloadable, safe to delete).</li>'
f'</ul>'
f'<p><b>Back up <code>~/.tinyweb/</code> periodically.</b> '
f'Copying the whole directory to another device preserves your identity and index together. '
f'The <a href="/export">export</a> page gives you a JSON dump of pages only — '
f'it does not preserve your identity or subscription state, so it is a migration aid, '
f'not a substitute for a full backup.</p>'
f'<h2>what is the slow web?</h2>'
f'<p>The slow web is a movement for intentionality over speed, '
f'human curation over algorithmic feeds, privacy over surveillance, '
@ -851,7 +1090,7 @@ def handle_tag_browse(tag_name, query=None):
tag_links = " ".join(f'<a href="/tags/{esc(t)}">[{esc(t)}]</a>' for t in tags)
items += (
f'<li>{esc(r["title"])}{note_html} {tag_links} '
f'<small>(<a href="{esc(r["url"])}">{esc(r["url"])}</a>)</small></li>'
f'<small>(<a href="{esc(r["url"])}" rel="noreferrer noopener">{esc(r["url"])}</a>)</small></li>'
)
finally:
return_db(db)
@ -864,6 +1103,125 @@ def handle_tag_browse(tag_name, query=None):
)
MAX_API_SITES = 5000
def _page_is_shared(tags, mode):
"""Decide whether a page with the given tags is shared under the given mode.
`private` always wins a page tagged private is never shared, regardless of mode.
"""
if "private" in tags:
return False
if mode == "require_public" and "public" not in tags:
return False
return True
def _shared_sites(db, since=""):
"""Return the full site records that a subscriber would receive.
The caller owns the db connection.
"""
mode = get_setting("sharing_mode", "exclude_private")
if since:
rows = db.execute(
"SELECT id, url, title, note, last_modified FROM pages "
"WHERE last_modified > ? ORDER BY id DESC LIMIT ?",
(since, MAX_API_SITES),
).fetchall()
else:
rows = db.execute(
"SELECT id, url, title, note, last_modified FROM pages ORDER BY id DESC LIMIT ?",
(MAX_API_SITES,),
).fetchall()
sites = []
for r in rows:
tags = _get_page_tags(r["id"], db)
if not _page_is_shared(tags, mode):
continue
sites.append({
"url": r["url"], "title": r["title"], "note": r["note"],
"tags": tags, "last_modified": r["last_modified"] or "",
})
return sites
def _shared_all_urls(db):
"""Return the URL list a subscriber uses to detect deletions."""
mode = get_setting("sharing_mode", "exclude_private")
rows = db.execute(
"SELECT id, url FROM pages ORDER BY id DESC LIMIT ?", (MAX_API_SITES,)
).fetchall()
return [r["url"] for r in rows if _page_is_shared(_get_page_tags(r["id"], db), mode)]
def _count_shared_pages():
"""Cheap page count under the current sharing rule — used by the settings UI."""
db = get_db()
try:
return len(_shared_all_urls(db))
finally:
return_db(db)
def handle_share_preview():
"""Show the list of pages a subscriber would currently receive.
Works regardless of whether sharing is enabled lets the user see the surface
before flipping it on.
"""
mode = get_setting("sharing_mode", "exclude_private")
mode_label = (
"only pages tagged <code>public</code>"
if mode == "require_public"
else "all pages except those tagged <code>private</code>"
)
sharing_on = get_setting("sharing_enabled", "0") == "1"
status = (
'<p>Sharing is <b>enabled</b>. Subscribers see the pages listed below.</p>'
if sharing_on else
'<p>Sharing is <b>disabled</b>. Nothing is actually being shared right now; '
'this is the list that would be exposed if you enabled it.</p>'
)
db = get_db()
try:
sites = _shared_sites(db)
finally:
return_db(db)
if not sites:
body = (
"<h1>sharing preview</h1>"
f"<p>Rule: {mode_label}.</p>"
f"{status}"
"<p><em>No pages match the current rule.</em></p>"
'<p><a href="/style">back to settings</a></p>'
)
return _respond(body)
rows = ""
for s in sites:
tags_html = ""
if s["tags"]:
tags_html = " " + " ".join(f"[{esc(t)}]" for t in s["tags"])
note_html = f' — <em>{esc(s["note"])}</em>' if s["note"] else ""
rows += (
f'<li>'
f'<a href="{esc(s["url"])}" rel="noreferrer noopener">{esc(s["title"] or s["url"])}</a>'
f'{note_html}{tags_html} '
f'<br><small>{esc(s["url"])}</small>'
f'</li>'
)
body = (
"<h1>sharing preview</h1>"
f"<p>Rule: {mode_label}.</p>"
f"{status}"
f"<p><b>{len(sites)}</b> page(s) visible to subscribers.</p>"
f"<ul>{rows}</ul>"
'<p><a href="/style">back to settings</a></p>'
)
return _respond(body)
def handle_api_sites(query=None):
if get_setting("sharing_enabled", "0") != "1":
return _json_response(
@ -874,23 +1232,8 @@ def handle_api_sites(query=None):
since = (query or {}).get("since", [""])[0].strip()
db = get_db()
try:
if since:
rows = db.execute(
"SELECT id, url, title, note, last_modified FROM pages "
"WHERE last_modified > ? ORDER BY id DESC",
(since,),
).fetchall()
else:
rows = db.execute("SELECT id, url, title, note, last_modified FROM pages ORDER BY id DESC").fetchall()
sites = []
for r in rows:
tags = _get_page_tags(r["id"], db)
sites.append({
"url": r["url"], "title": r["title"], "note": r["note"],
"tags": tags, "last_modified": r["last_modified"] or "",
})
# Include list of all current URLs so subscriber can detect deletions
all_urls = [r["url"] for r in db.execute("SELECT url FROM pages").fetchall()] if not since else None
sites = _shared_sites(db, since=since)
all_urls = _shared_all_urls(db) if not since else None
finally:
return_db(db)
data = {"name": get_site_name(), "sites": sites}
@ -899,6 +1242,9 @@ def handle_api_sites(query=None):
return _json_response(data, headers={"Access-Control-Allow-Origin": "*"})
_sync_threads = {}
def handle_subscriptions(msg=""):
db = get_db()
try:
@ -907,30 +1253,54 @@ def handle_subscriptions(msg=""):
return_db(db)
cards = ""
for s in subs:
sub_id = s["id"]
auto_label = "on" if s["auto_sync"] else "off"
last = s["last_sync"] or "never"
sync_status = get_setting(f"sync_status_{sub_id}", "")
is_syncing = sub_id in _sync_threads and _sync_threads[sub_id].is_alive()
# Status line: show syncing indicator or last result
if is_syncing:
status_html = '<div style="margin-top:0.4rem;font-size:0.85rem;color:#2070c0">syncing...</div>'
elif sync_status.startswith("error:"):
err_msg = sync_status[6:]
status_html = f'<div style="margin-top:0.4rem;font-size:0.85rem;color:#c03030">{esc(err_msg)}</div>'
else:
status_html = ""
# Disable sync button while syncing
if is_syncing:
sync_btn = '<button disabled>syncing...</button>'
else:
sync_btn = (
f'<form method="post" action="/subscriptions/sync/{sub_id}" style="display:inline">'
f'{_csrf_field()}<button>sync now</button></form>'
)
cards += (
f'<div style="border:1px solid #ddd;border-radius:4px;padding:0.9rem 1rem;margin-bottom:0.75rem">'
f'<div style="margin-bottom:0.4rem"><b>{esc(s["name"] or "unknown")}</b></div>'
f'<div><small>{esc(s["dest_hash"])}</small></div>'
f'<div style="margin-top:0.4rem;font-size:0.85rem;color:#606060">last sync: {esc(last)}</div>'
f'{status_html}'
f'<div style="display:flex;gap:0.5rem;align-items:center;flex-wrap:wrap;margin-top:0.7rem">'
f'<a href="/subscriptions/browse/{s["id"]}">browse</a>'
f'<form method="post" action="/subscriptions/sync/{s["id"]}" style="display:inline">'
f'{_csrf_field()}<button>sync now</button></form>'
f'<form method="post" action="/subscriptions/autosync/{s["id"]}" style="display:inline">'
f'<a href="/subscriptions/browse/{sub_id}">browse</a>'
f'{sync_btn}'
f'<form method="post" action="/subscriptions/autosync/{sub_id}" style="display:inline">'
f'{_csrf_field()}<button>auto-sync: {auto_label}</button></form>'
f'<form method="post" action="/subscriptions/delete/{s["id"]}" style="display:inline">'
f'<form method="post" action="/subscriptions/delete/{sub_id}" style="display:inline">'
f'{_csrf_field()}<button>remove</button></form>'
f'</div>'
f'</div>'
)
listing = ""
if subs:
any_syncing = any(sid in _sync_threads and _sync_threads[sid].is_alive() for sid in [s["id"] for s in subs])
syncall_btn = '<button disabled>syncing...</button>' if any_syncing else '<button>sync all</button>'
listing = (
f'{cards}'
f'<form method="post" action="/subscriptions/syncall">'
f'{_csrf_field()}<button>sync all</button></form>'
f'{_csrf_field()}{syncall_btn}</form>'
)
return _respond(
f"<h1>subscriptions</h1>"
@ -974,18 +1344,20 @@ def handle_subscription_add(body):
return handle_subscriptions(f"Subscribed to {esc(name or dest_hash)}.")
MAX_BROWSE = 5000
def handle_subscription_browse(sub_id):
db = get_db()
try:
sub = db.execute("SELECT * FROM subscriptions WHERE id = ?", (sub_id,)).fetchone()
if not sub:
return _error(404)
local_urls = set(r["url"] for r in db.execute("SELECT url FROM pages").fetchall())
local_urls = set(r["url"] for r in db.execute("SELECT url FROM pages LIMIT ?", (MAX_BROWSE,)).fetchall())
# Use locally synced data if available, otherwise fetch live
remote_rows = db.execute(
"SELECT url, title, note, tags FROM remote_pages WHERE subscription_id = ?",
(sub_id,),
"SELECT url, title, note, tags FROM remote_pages WHERE subscription_id = ? LIMIT ?",
(sub_id, MAX_BROWSE),
).fetchall()
finally:
return_db(db)
@ -1055,7 +1427,7 @@ def handle_subscription_pick(body):
remote_tags = {r["url"]: r["tags"] for r in remote_rows}
if import_all:
local_urls = set(r["url"] for r in db.execute("SELECT url FROM pages").fetchall())
local_urls = set(r["url"] for r in db.execute("SELECT url FROM pages LIMIT ?", (MAX_BROWSE,)).fetchall())
urls = [r["url"] for r in remote_rows if r["url"] not in local_urls]
else:
urls = body.get("urls", [])
@ -1087,13 +1459,15 @@ def handle_subscription_pick(body):
return handle_subscriptions(f"Imported {imported} page(s). {errors} error(s).")
def handle_subscription_sync(sub_id):
def _sync_subscription(sub_id):
"""Run a single subscription sync. Designed to run in a background thread."""
set_setting(f"sync_status_{sub_id}", "syncing")
db = get_db()
try:
sub = db.execute("SELECT * FROM subscriptions WHERE id = ?", (sub_id,)).fetchone()
if not sub:
return handle_subscriptions("Subscription not found.")
# Use last_sync for delta sync if available
set_setting(f"sync_status_{sub_id}", "error:Subscription not found.")
return
since = sub["last_sync"].replace(" ", "T") if sub["last_sync"] else ""
try:
data = fetch_remote_sites(sub["dest_hash"], since=since)
@ -1101,11 +1475,12 @@ def handle_subscription_sync(sub_id):
all_urls = data.get("all_urls")
remote_name = data.get("name", sub["name"])
except PermissionError:
return handle_subscriptions("That instance has sharing disabled.")
except Exception:
return handle_subscriptions("Could not sync with that instance.")
set_setting(f"sync_status_{sub_id}", "error:That instance has sharing disabled.")
return
except Exception as e:
set_setting(f"sync_status_{sub_id}", f"error:Could not sync \u2014 {e}")
return
# If full sync (all_urls provided), remove pages no longer on remote
if all_urls is not None:
existing = db.execute(
"SELECT id, url FROM remote_pages WHERE subscription_id = ?", (sub_id,)
@ -1115,7 +1490,6 @@ def handle_subscription_sync(sub_id):
if row["url"] not in remote_url_set:
db.execute("DELETE FROM remote_pages WHERE id = ?", (row["id"],))
# Upsert changed/new pages
synced = 0
for s in sites:
try:
@ -1125,8 +1499,7 @@ def handle_subscription_sync(sub_id):
"ON CONFLICT(subscription_id, url) DO UPDATE SET title=excluded.title, note=excluded.note, tags=excluded.tags",
(sub_id, s["url"], s["title"], s.get("note", ""), tags_str),
)
# Embed remote page for semantic search
if get_setting("semantic_search", "1") == "1":
if get_setting("semantic_search", "0") == "1":
try:
from embeddings import store_remote_embeddings
rp_id = db.execute(
@ -1142,9 +1515,22 @@ def handle_subscription_sync(sub_id):
now = datetime.now().strftime("%Y-%m-%dT%H:%M:%S")
db.execute("UPDATE subscriptions SET last_sync = ?, name = ? WHERE id = ?", (now, remote_name, sub_id))
db.commit()
set_setting(f"sync_status_{sub_id}", f"done:{synced}")
except Exception as e:
set_setting(f"sync_status_{sub_id}", f"error:{e}")
finally:
return_db(db)
return handle_subscriptions(f"Synced {synced} site(s) from {esc(remote_name)}.")
def handle_subscription_sync(sub_id):
if sub_id in _sync_threads and _sync_threads[sub_id].is_alive():
return _redirect("/subscriptions")
# Clear previous status
set_setting(f"sync_status_{sub_id}", "syncing")
t = threading.Thread(target=_sync_subscription, args=(sub_id,), daemon=True)
_sync_threads[sub_id] = t
t.start()
return _redirect("/subscriptions")
def handle_subscription_autosync(sub_id):
@ -1176,53 +1562,15 @@ def handle_subscription_syncall():
return_db(db)
if not subs:
return handle_subscriptions("No subscriptions have auto-sync enabled.")
total = 0
for sub in subs:
try:
since = sub["last_sync"].replace(" ", "T") if sub["last_sync"] else ""
data = fetch_remote_sites(sub["dest_hash"], since=since)
sites = data.get("sites", [])
all_urls = data.get("all_urls")
remote_name = data.get("name", sub["name"])
db = get_db()
try:
if all_urls is not None:
existing = db.execute(
"SELECT id, url FROM remote_pages WHERE subscription_id = ?", (sub["id"],)
).fetchall()
remote_url_set = set(all_urls)
for row in existing:
if row["url"] not in remote_url_set:
db.execute("DELETE FROM remote_pages WHERE id = ?", (row["id"],))
for s in sites:
try:
tags_str = ",".join(s.get("tags", []))
db.execute(
"INSERT INTO remote_pages (subscription_id, url, title, note, tags) VALUES (?, ?, ?, ?, ?) "
"ON CONFLICT(subscription_id, url) DO UPDATE SET title=excluded.title, note=excluded.note, tags=excluded.tags",
(sub["id"], s["url"], s["title"], s.get("note", ""), tags_str),
)
if get_setting("semantic_search", "1") == "1":
try:
from embeddings import store_remote_embeddings
rp_id = db.execute(
"SELECT id FROM remote_pages WHERE subscription_id = ? AND url = ?",
(sub["id"], s["url"]),
).fetchone()["id"]
store_remote_embeddings(rp_id, s["title"], s.get("note", ""), db)
except Exception:
pass
except Exception:
pass
now = datetime.now().strftime("%Y-%m-%dT%H:%M:%S")
db.execute("UPDATE subscriptions SET last_sync = ?, name = ? WHERE id = ?", (now, remote_name, sub["id"]))
db.commit()
finally:
return_db(db)
total += 1
except Exception:
pass
return handle_subscriptions(f"Synced {total} subscription(s).")
sub_id = sub["id"]
if sub_id in _sync_threads and _sync_threads[sub_id].is_alive():
continue
set_setting(f"sync_status_{sub_id}", "syncing")
t = threading.Thread(target=_sync_subscription, args=(sub_id,), daemon=True)
_sync_threads[sub_id] = t
t.start()
return _redirect("/subscriptions")
# --- Reindex (semantic search) ---
@ -1232,7 +1580,7 @@ _reindex_thread = None
def handle_reindex_form():
if get_setting("semantic_search", "1") != "1":
if get_setting("semantic_search", "0") != "1":
return _respond(
f"<h2>semantic search index</h2>"
f"<p>Semantic search is disabled. Enable it in <a href=\"/style\">settings</a> to use embeddings.</p>"
@ -1319,10 +1667,12 @@ def _dispatch_inner(data):
return handle_bookmark(query)
elif path == "/style":
return handle_style_form()
elif path == "/share/preview":
return handle_share_preview()
elif path == "/about":
return handle_about()
elif path == "/export":
return handle_export()
return handle_export(query)
elif path == "/import":
return handle_import_form()
elif path == "/tags":
@ -1344,6 +1694,8 @@ def _dispatch_inner(data):
return _respond("<h1>403 Forbidden</h1><p>Invalid or missing CSRF token.</p>", status=403)
if path == "/add":
return handle_add_submit(body)
elif path == "/pages/bulk":
return handle_bulk_action(body)
elif path == "/add/manual":
return handle_add_manual_submit(body)
elif path.startswith("/edit/"):
@ -1357,6 +1709,10 @@ def _dispatch_inner(data):
elif path == "/style/reset":
set_setting("custom_template", "")
return handle_style_form("Template reset to default.")
elif path == "/style/vacuum":
from db import vacuum_db
vacuum_db()
return handle_style_form("Database vacuumed.")
elif path == "/import":
return handle_import_submit(body)
elif path == "/reindex":
@ -1397,8 +1753,7 @@ def dispatch_request(data):
resp["headers"]["Content-Security-Policy"] = (
"default-src 'self'; "
"script-src 'self' 'unsafe-inline'; "
"style-src 'self' 'unsafe-inline' https://fonts.googleapis.com; "
"font-src 'self' https://fonts.gstatic.com; "
"style-src 'self' 'unsafe-inline'; "
"img-src * data:; "
"frame-ancestors 'none'; "
"form-action 'self'; "

5
pytest.ini Normal file
View file

@ -0,0 +1,5 @@
[pytest]
testpaths = tests
python_files = test_*.py
filterwarnings =
ignore::DeprecationWarning

2
requirements-dev.txt Normal file
View file

@ -0,0 +1,2 @@
-r requirements.txt
pytest

View file

@ -1,9 +1,15 @@
import json
import time
import RNS
APP_NAME = "tinyweb"
ASPECTS = ["server"]
REQUEST_TIMEOUT = 30
# Two-tier timeout profiles: fast first, then slow for LoRa/multi-hop links
_TIMEOUT_TIERS = [
{"path": 15, "link": 15, "request": 30, "poll": 0.25},
{"path": 60, "link": 60, "request": 120, "poll": 1.0},
]
def fetch_remote_sites(dest_hash_hex, since=""):
@ -11,18 +17,40 @@ def fetch_remote_sites(dest_hash_hex, since=""):
Connect to a remote TinyWeb instance over Reticulum and fetch its
shared sites. Returns the response dict from /api/sites, or raises
an exception on failure. Pass `since` as ISO timestamp for delta sync.
Uses progressive timeouts: tries fast first, then retries with longer
timeouts for slow links (LoRa, multi-hop).
"""
last_error = None
for tier in _TIMEOUT_TIERS:
try:
return _fetch(dest_hash_hex, since, tier)
except PermissionError:
raise # Don't retry permission errors
except Exception as e:
last_error = e
continue
raise ConnectionError(
f"Could not reach {dest_hash_hex} after {len(_TIMEOUT_TIERS)} attempts: {last_error}"
)
def _fetch(dest_hash_hex, since, timeouts):
"""Single fetch attempt with the given timeout profile."""
dest_hash = bytes.fromhex(dest_hash_hex)
poll = timeouts["poll"]
# Resolve path if needed
if not RNS.Transport.has_path(dest_hash):
RNS.Transport.request_path(dest_hash)
elapsed = 0
while not RNS.Transport.has_path(dest_hash) and elapsed < 15:
time.sleep(0.5)
elapsed += 0.5
while not RNS.Transport.has_path(dest_hash) and elapsed < timeouts["path"]:
time.sleep(poll)
elapsed += poll
if not RNS.Transport.has_path(dest_hash):
raise ConnectionError(f"Could not find path to {dest_hash_hex}")
raise ConnectionError(
f"Could not find path to {dest_hash_hex} ({timeouts['path']}s timeout)"
)
server_identity = RNS.Identity.recall(dest_hash)
if server_identity is None:
@ -39,15 +67,16 @@ def fetch_remote_sites(dest_hash_hex, since=""):
# Establish link
link = RNS.Link(destination)
elapsed = 0
while link.status == RNS.Link.PENDING and elapsed < 15:
time.sleep(0.25)
elapsed += 0.25
while link.status == RNS.Link.PENDING and elapsed < timeouts["link"]:
time.sleep(poll)
elapsed += poll
if link.status != RNS.Link.ACTIVE:
raise ConnectionError(f"Could not establish link to {dest_hash_hex}")
raise ConnectionError(
f"Could not establish link to {dest_hash_hex} ({timeouts['link']}s timeout)"
)
try:
# Request /api/sites
query = {"since": [since]} if since else {}
request_data = {
"method": "GET",
@ -57,13 +86,14 @@ def fetch_remote_sites(dest_hash_hex, since=""):
"gateway_host": "",
}
receipt = link.request("/tinyweb", data=request_data, timeout=REQUEST_TIMEOUT)
req_timeout = timeouts["request"]
receipt = link.request("/tinyweb", data=request_data, timeout=req_timeout)
elapsed = 0
done = (RNS.RequestReceipt.READY, RNS.RequestReceipt.DELIVERED, RNS.RequestReceipt.FAILED)
while receipt.get_status() not in done and elapsed < REQUEST_TIMEOUT:
time.sleep(0.5)
elapsed += 0.5
while receipt.get_status() not in done and elapsed < req_timeout:
time.sleep(poll)
elapsed += poll
if receipt.get_status() in (RNS.RequestReceipt.READY, RNS.RequestReceipt.DELIVERED):
resp = receipt.get_response()
@ -71,9 +101,10 @@ def fetch_remote_sites(dest_hash_hex, since=""):
raise PermissionError("That instance has sharing disabled.")
if resp["status"] != 200:
raise ConnectionError(f"Remote returned status {resp['status']}")
import json
return json.loads(resp["body"])
else:
raise ConnectionError(f"Request failed or timed out")
raise ConnectionError(
f"Request failed or timed out ({req_timeout}s timeout)"
)
finally:
link.teardown()

View file

@ -7,13 +7,13 @@ def esc(s):
DEFAULT_TEMPLATE = "<html>\n<head>\n</head>\n<body>\n{{content}}\n</body>\n</html>"
DEFAULT_TEMPLATE = "<html>\n<head>\n<meta name=\"referrer\" content=\"no-referrer\">\n<meta http-equiv=\"x-dns-prefetch-control\" content=\"off\">\n</head>\n<body>\n{{content}}\n</body>\n</html>"
def _default_template():
name = esc(get_setting("site_name", "tinyweb"))
return (
"<html>\n<head>\n</head>\n<body>\n"
'<html>\n<head>\n<meta name="referrer" content="no-referrer">\n<meta http-equiv="x-dns-prefetch-control" content="off">\n</head>\n<body>\n'
f'<p><b><a href="/">{name}</a></b>'
' | <a href="/">search</a> | <a href="/pages">browse</a>'
' | <a href="/tags">tags</a> | <a href="/subscriptions">subscriptions</a>'

60
tests/test_csrf.py Normal file
View file

@ -0,0 +1,60 @@
"""Tests for `_check_csrf` — form-submission CSRF protection.
Every POST handler calls this to verify the submitted _csrf field matches
the token stored in the thread-local (which is seeded from the cookie by
`dispatch_request`). Missing or mismatched tokens must fail closed.
"""
import handlers as handlers_module
from handlers import _check_csrf, _csrf_field, _get_csrf_token
def _set_token(token):
handlers_module._request_local.csrf_token = token
def _clear_token():
if hasattr(handlers_module._request_local, "csrf_token"):
del handlers_module._request_local.csrf_token
def teardown_function(_):
_clear_token()
def test_rejects_missing_token_in_body():
_set_token("server-side-token")
assert _check_csrf({}) is False
def test_rejects_empty_token_in_body():
_set_token("server-side-token")
assert _check_csrf({"_csrf": [""]}) is False
def test_rejects_mismatched_token():
_set_token("server-side-token")
assert _check_csrf({"_csrf": ["attacker-token"]}) is False
def test_accepts_matching_token():
_set_token("server-side-token")
assert _check_csrf({"_csrf": ["server-side-token"]}) is True
def test_rejects_when_server_token_missing():
"""If the server-side token is empty (shouldn't happen after dispatch_request
seeds it, but be defensive), the check must fail closed."""
_clear_token()
assert _check_csrf({"_csrf": ["anything"]}) is False
def test_csrf_field_renders_current_token():
_set_token("abc123")
field = _csrf_field()
assert 'name="_csrf"' in field
assert 'value="abc123"' in field
def test_get_csrf_token_returns_empty_when_unset():
_clear_token()
assert _get_csrf_token() == ""

155
tests/test_db_index_url.py Normal file
View file

@ -0,0 +1,155 @@
"""Tests for `index_url` — the main write path.
Covers UPSERT behavior, links being replaced on re-index, FTS index staying
in sync via triggers, and the connection pool returning clean connections.
"""
from unittest.mock import patch
from conftest import patch_dns_ok
import db as db_module
from db import get_db, return_db, index_url
def _mock_fetch_page(title="Test Page", body="test body text", links=None, meta=""):
"""Return a replacement for db.fetch_page that yields canned data."""
links = links or []
def fake(url):
return (title, body, links, meta)
return fake
def test_insert_creates_page_row_and_fts_entry(temp_db, monkeypatch):
patch_dns_ok(monkeypatch)
monkeypatch.setattr(db_module, "fetch_page", _mock_fetch_page(
title="Rust Intro", body="ownership and borrowing basics", links=[],
))
index_url("https://example.com/rust")
db = get_db()
try:
row = db.execute("SELECT id, title, body FROM pages").fetchone()
assert row is not None
assert row["title"] == "Rust Intro"
assert "ownership" in row["body"]
# Verify FTS trigger fired.
fts_hits = db.execute(
"SELECT rowid FROM pages_fts WHERE pages_fts MATCH 'ownership*'"
).fetchall()
assert len(fts_hits) == 1
assert fts_hits[0]["rowid"] == row["id"]
finally:
return_db(db)
def test_re_indexing_same_url_updates_in_place(temp_db, monkeypatch):
patch_dns_ok(monkeypatch)
monkeypatch.setattr(db_module, "fetch_page", _mock_fetch_page(
title="First Title", body="first body", links=[],
))
index_url("https://example.com/page")
monkeypatch.setattr(db_module, "fetch_page", _mock_fetch_page(
title="Second Title", body="second body", links=[],
))
index_url("https://example.com/page")
db = get_db()
try:
rows = db.execute("SELECT title, body FROM pages").fetchall()
finally:
return_db(db)
assert len(rows) == 1, "re-indexing should UPDATE not INSERT"
assert rows[0]["title"] == "Second Title"
def test_links_replaced_on_reindex(temp_db, monkeypatch):
patch_dns_ok(monkeypatch)
monkeypatch.setattr(db_module, "fetch_page", _mock_fetch_page(
title="T", body="b",
links=[("https://example.com/a", "first"), ("https://example.com/b", "second")],
))
index_url("https://example.com/src")
monkeypatch.setattr(db_module, "fetch_page", _mock_fetch_page(
title="T", body="b",
links=[("https://example.com/c", "third-only")],
))
index_url("https://example.com/src")
db = get_db()
try:
rows = db.execute("SELECT url FROM links").fetchall()
finally:
return_db(db)
urls = {r["url"] for r in rows}
assert urls == {"https://example.com/c"}, "old links should be deleted on reindex"
def test_url_cleaned_before_insert(temp_db, monkeypatch):
"""index_url should apply clean_url before touching the DB, so tracking params
don't create duplicate rows."""
patch_dns_ok(monkeypatch)
monkeypatch.setattr(db_module, "fetch_page", _mock_fetch_page(title="T", body="b"))
index_url("https://example.com/page?utm_source=twitter#frag")
db = get_db()
try:
rows = db.execute("SELECT url FROM pages").fetchall()
finally:
return_db(db)
assert len(rows) == 1
assert rows[0]["url"] == "https://example.com/page"
def test_summary_populated_from_meta_description(temp_db, monkeypatch):
patch_dns_ok(monkeypatch)
long_meta = "A thoughtful description that exceeds twenty chars"
monkeypatch.setattr(db_module, "fetch_page", _mock_fetch_page(
title="T", body="b", meta=long_meta,
))
index_url("https://example.com/page")
db = get_db()
try:
row = db.execute("SELECT summary FROM pages").fetchone()
finally:
return_db(db)
assert row["summary"] == long_meta
def test_short_meta_description_not_stored_as_summary(temp_db, monkeypatch):
patch_dns_ok(monkeypatch)
monkeypatch.setattr(db_module, "fetch_page", _mock_fetch_page(
title="T", body="b", meta="too short",
))
index_url("https://example.com/page")
db = get_db()
try:
row = db.execute("SELECT summary FROM pages").fetchone()
finally:
return_db(db)
assert row["summary"] == ""
def test_pool_returns_clean_connection(temp_db, monkeypatch):
"""Regression for 1bc695f — `return_db` should roll back uncommitted work
so the next consumer doesn't see stale state."""
patch_dns_ok(monkeypatch)
monkeypatch.setattr(db_module, "fetch_page", _mock_fetch_page(title="T", body="b"))
index_url("https://example.com/one")
# Take a connection, make a dirty uncommitted change, return it.
db = get_db()
db.execute("INSERT INTO pages (url, title, body) VALUES (?, ?, ?)",
("https://dirty.example.com/", "dirty", "dirty"))
# NOTE: no commit here — this is the dirty state we want rolled back.
return_db(db)
# A later consumer must not see the dirty row.
db2 = get_db()
try:
urls = {r["url"] for r in db2.execute("SELECT url FROM pages").fetchall()}
finally:
return_db(db2)
assert "https://dirty.example.com/" not in urls

90
tests/test_db_schema.py Normal file
View file

@ -0,0 +1,90 @@
"""Tests for `init_db` and the settings key-value store.
`init_db` is called unconditionally on startup, so it must be idempotent
and create every table/trigger the rest of the app expects.
"""
from db import get_db, return_db, init_db, get_setting, set_setting, get_site_name
EXPECTED_TABLES = {
"pages", "links", "settings", "subscriptions",
"remote_pages", "tags", "page_tags", "chunks",
# FTS5 virtual tables:
"pages_fts", "remote_pages_fts",
}
def test_all_expected_tables_exist(temp_db):
db = get_db()
try:
rows = db.execute(
"SELECT name FROM sqlite_master WHERE type IN ('table') AND name NOT LIKE 'sqlite_%'"
).fetchall()
names = {r["name"] for r in rows}
finally:
return_db(db)
missing = EXPECTED_TABLES - names
assert not missing, f"tables missing after init_db: {missing}"
def test_fts_triggers_exist(temp_db):
db = get_db()
try:
rows = db.execute(
"SELECT name FROM sqlite_master WHERE type = 'trigger'"
).fetchall()
names = {r["name"] for r in rows}
finally:
return_db(db)
# These triggers keep pages_fts in sync with pages on insert/update/delete.
for trigger in ("pages_ai", "pages_ad", "pages_au"):
assert trigger in names, f"missing trigger {trigger}"
def test_init_db_is_idempotent(temp_db):
"""Running init_db twice on the same DB must not error or duplicate anything."""
init_db()
init_db() # second call should be a no-op
db = get_db()
try:
count = db.execute(
"SELECT count(*) FROM sqlite_master WHERE name = 'pages'"
).fetchone()[0]
finally:
return_db(db)
assert count == 1
def test_get_setting_returns_default_when_missing(temp_db):
assert get_setting("nonexistent", "fallback") == "fallback"
assert get_setting("nonexistent") == ""
def test_set_setting_then_get(temp_db):
set_setting("site_name", "my-personal-index")
assert get_setting("site_name") == "my-personal-index"
def test_set_setting_updates_existing(temp_db):
set_setting("key", "first")
set_setting("key", "second")
assert get_setting("key") == "second"
def test_get_site_name_has_default(temp_db):
assert get_site_name() == "tinyweb"
def test_get_site_name_reflects_override(temp_db):
set_setting("site_name", "custom-site")
assert get_site_name() == "custom-site"
def test_foreign_keys_pragma_enabled(temp_db):
"""Pool connections should have foreign_keys=ON so CASCADE deletes work."""
db = get_db()
try:
row = db.execute("PRAGMA foreign_keys").fetchone()
finally:
return_db(db)
assert row[0] == 1

113
tests/test_fts_sanitizer.py Normal file
View file

@ -0,0 +1,113 @@
"""Tests for `_sanitize_fts_query`.
The sanitizer is the boundary between user input and FTS5 MATCH syntax.
Commit 1bc695f tightened it after noticing that colons and operator words
could escape the quoting. These tests keep that regression dead.
"""
import pytest
from handlers import _sanitize_fts_query
def test_empty_query_returns_no_match_token():
assert _sanitize_fts_query("") == '""'
assert _sanitize_fts_query(" ") == '""'
def test_single_word_becomes_prefix_match():
assert _sanitize_fts_query("rust") == "rust*"
def test_multi_word_quotes_all_but_last():
result = _sanitize_fts_query("rust borrow checker")
assert result == '"rust" "borrow" checker*'
def test_stopwords_are_dropped():
# "the" and "a" should vanish; only "cat" remains (and gets prefix star).
assert _sanitize_fts_query("the a cat") == "cat*"
def test_all_stopwords_returns_no_match_token():
assert _sanitize_fts_query("the and or") == '""'
@pytest.mark.parametrize("bad_char", ["'", "(", ")", "+", "-", "^", "~", ":"])
def test_fts5_operators_stripped_from_tokens(bad_char):
"""FTS5 special chars inside user tokens must not survive — regression for 1bc695f.
The sanitizer legitimately adds `"` around tokens and a trailing `*` for prefix
matching; both are excluded from this check.
"""
payload = f"foo{bad_char}bar"
out = _sanitize_fts_query(payload)
assert bad_char not in out, f"{bad_char!r} leaked into {out!r}"
def test_asterisk_only_appears_as_trailing_prefix():
"""Input `*` should not become an in-token asterisk; the sanitizer's trailing `*` is fine."""
out = _sanitize_fts_query("foo*bar")
assert out.count("*") <= 1
if "*" in out:
assert out.endswith("*")
def test_quote_in_input_does_not_break_out_of_quoted_token():
"""A `"` in user input must not close the sanitizer's protective quoting.
The sanitizer wraps each non-last token in double quotes; if a stray `"` from
the user slipped through, the resulting FTS5 expression would be interpreted
as broken syntax or, worse, a column filter.
"""
out = _sanitize_fts_query('foo"bar baz"qux')
# Each pair of quotes in the output should be balanced and around a clean token.
assert out.count('"') % 2 == 0
# No embedded quotes inside a quoted region.
import re
for match in re.findall(r'"[^"]*"', out):
inner = match[1:-1]
assert '"' not in inner
@pytest.mark.parametrize("op", ["AND", "OR", "NOT", "NEAR", "and", "or", "not", "near"])
def test_fts5_operator_words_dropped(op):
"""AND/OR/NOT/NEAR would be interpreted as operators on the unquoted last token."""
out = _sanitize_fts_query(f"foo {op} bar")
# the operator word itself should not appear
assert op.upper() not in out.upper().split('"'), f"operator {op!r} survived in {out!r}"
def test_injection_payload_produces_valid_fts5():
"""End-to-end: a realistic injection payload must produce syntactically valid FTS5.
We run the sanitized output through a throwaway FTS5 table; if the sanitizer
leaks operator characters the MATCH either raises or interprets malicious syntax.
"""
import sqlite3
conn = sqlite3.connect(":memory:")
conn.execute("CREATE VIRTUAL TABLE t USING fts5(body)")
conn.execute("INSERT INTO t (body) VALUES ('hello world')")
for payload in [
'foo": OR bar NOT baz AND qux*()',
'" OR 1=1 --',
"title:secret AND public",
"(((",
"^^^~~~",
]:
q = _sanitize_fts_query(payload)
# Must not raise — if operators leaked, FTS5 would error or mis-parse.
conn.execute("SELECT * FROM t WHERE t MATCH ?", (q,)).fetchall()
conn.close()
def test_whitespace_only_tokens_dropped():
# tokens that become empty after stripping special chars should not produce bare quotes
out = _sanitize_fts_query('""" "" ""')
assert out == '""'
def test_colon_stripped():
"""Regression for 1bc695f — colon is an FTS5 column filter and must be stripped."""
out = _sanitize_fts_query("title:secret")
assert ":" not in out

View file

@ -0,0 +1,164 @@
"""Tests for gateway-level guards: body-size cap and Reticulum surface whitelist.
Regression targets from commit 1bc695f a 16 MiB upload limit (DoS guard)
and a strict GET-/api/sites-only whitelist for requests arriving over the
Reticulum mesh (CSRF can't protect mesh callers, so gate by whitelist).
"""
import io
import pytest
import app as app_module
from gateway import GatewayHandler, MAX_BODY_SIZE
class FakeHeaders:
"""Minimal replacement for http.server request headers."""
def __init__(self, items=None):
self._items = dict(items or {})
def get(self, key, default=None):
return self._items.get(key, default)
class FakeGatewayHandler(GatewayHandler):
"""Bypass the socket-bound __init__ and capture response calls in memory."""
def __init__(self, path="/", method="POST", headers=None, rfile=None):
self.path = path
self.command = method
self.headers = FakeHeaders(headers or {})
self.rfile = rfile or io.BytesIO()
self.wfile = io.BytesIO()
self._captured = {
"error": None, "status": None, "headers": [], "body_written": None,
}
def send_error(self, code, msg=""):
self._captured["error"] = (code, msg)
def send_response(self, code):
self._captured["status"] = code
def send_header(self, k, v):
self._captured["headers"].append((k, v))
def end_headers(self):
pass
def test_post_over_size_cap_rejected_with_413():
"""Regression for 1bc695f: request bodies over MAX_BODY_SIZE must be rejected
without being read into memory."""
oversize = MAX_BODY_SIZE + 1
handler = FakeGatewayHandler(
path="/add",
method="POST",
headers={"Content-Length": str(oversize)},
)
handler._forward("POST")
assert handler._captured["error"] is not None
code, _msg = handler._captured["error"]
assert code == 413
def test_post_at_size_cap_accepted():
"""A body exactly at MAX_BODY_SIZE should not be rejected by the size check."""
handler = FakeGatewayHandler(
path="/_does_not_matter",
method="POST",
headers={"Content-Length": str(MAX_BODY_SIZE)},
# rfile has no data; handler will try to read; local_dispatch isn't set.
# We only care that the 413 check passes, not that the request succeeds.
rfile=io.BytesIO(b""),
)
# Stub out local_dispatch so _forward doesn't try the network path.
from gateway import GatewayState
original = GatewayState.local_dispatch
GatewayState.local_dispatch = lambda data: {
"status": 404, "content_type": "text/plain", "body": "nope",
}
try:
handler._forward("POST")
finally:
GatewayState.local_dispatch = original
# Not a 413, because the body is exactly at the cap (cap is inclusive).
if handler._captured["error"]:
assert handler._captured["error"][0] != 413
def test_negative_content_length_rejected():
handler = FakeGatewayHandler(
path="/add",
method="POST",
headers={"Content-Length": "-1"},
)
handler._forward("POST")
assert handler._captured["error"] is not None
code, _msg = handler._captured["error"]
assert code == 400
def test_invalid_content_length_rejected():
handler = FakeGatewayHandler(
path="/add",
method="POST",
headers={"Content-Length": "abc"},
)
handler._forward("POST")
assert handler._captured["error"] is not None
code, _msg = handler._captured["error"]
assert code == 400
# -------- Reticulum mesh surface whitelist --------
def test_mesh_rejects_non_api_sites_get():
"""Regression for 1bc695f: remote mesh callers can only GET /api/sites."""
resp = app_module.rns_request_handler(
path="/tinyweb",
data={"method": "GET", "path": "/pages", "query": {}, "body": {}, "gateway_host": ""},
request_id="x", link_id="y", remote_identity=None, requested_at=0,
)
assert resp["status"] == 403
def test_mesh_rejects_post_to_api_sites():
resp = app_module.rns_request_handler(
path="/tinyweb",
data={"method": "POST", "path": "/api/sites", "query": {}, "body": {}, "gateway_host": ""},
request_id="x", link_id="y", remote_identity=None, requested_at=0,
)
assert resp["status"] == 403
def test_mesh_rejects_sensitive_local_endpoints():
for path in ("/add", "/delete/1", "/style", "/import", "/export"):
resp = app_module.rns_request_handler(
path="/tinyweb",
data={"method": "GET", "path": path, "query": {}, "body": {}, "gateway_host": ""},
request_id="x", link_id="y", remote_identity=None, requested_at=0,
)
assert resp["status"] == 403, f"path {path!r} leaked through mesh whitelist"
def test_mesh_allows_api_sites_get(temp_db, csrf_session):
"""Sanity check: the one whitelisted combination is accepted."""
resp = app_module.rns_request_handler(
path="/tinyweb",
data={"method": "GET", "path": "/api/sites", "query": {}, "body": {}, "gateway_host": ""},
request_id="x", link_id="y", remote_identity=None, requested_at=0,
)
# Status depends on handler output; 200 is the happy path.
assert resp["status"] in (200, 403) # 403 if sharing is disabled by default
def test_mesh_handles_missing_data_payload():
"""Regression-minded check: a None or malformed data object shouldn't crash."""
resp = app_module.rns_request_handler(
path="/tinyweb",
data=None,
request_id="x", link_id="y", remote_identity=None, requested_at=0,
)
# Default data has method=GET, path=/ which is not in the whitelist.
assert resp["status"] == 403

View file

@ -0,0 +1,174 @@
"""Tests for `handle_bulk_action`, edit flow, and the bulk-delete confirm step.
The bulk-delete confirmation flow is a data-loss guard added in commit
8dffd8c a stray POST without `confirmed=1` must render the confirmation
page instead of actually deleting.
"""
from db import get_db, return_db
from handlers import (
handle_bulk_action,
handle_edit_form,
handle_edit_submit,
handle_pages,
)
def _all_urls(seeded_db):
db = get_db()
try:
return {r["url"] for r in db.execute("SELECT url FROM pages").fetchall()}
finally:
return_db(db)
def _page_id(seeded_db, url):
db = get_db()
try:
return db.execute("SELECT id FROM pages WHERE url = ?", (url,)).fetchone()["id"]
finally:
return_db(db)
def test_bulk_delete_without_confirmed_renders_confirm_page(seeded_db, csrf_session):
"""Regression for 8dffd8c: bulk delete must NOT delete until confirmed=1 is set."""
pid = _page_id(seeded_db, "https://example.com/rust-intro")
urls_before = _all_urls(seeded_db)
resp = handle_bulk_action({
"ids": [str(pid)],
"action": ["delete"],
})
assert resp["status"] == 200
assert "confirm delete" in resp["body"].lower()
assert "Rust Intro" in resp["body"]
# Must still show a hidden confirmed=1 field in the follow-up form.
assert 'name="confirmed" value="1"' in resp["body"]
# Crucially: nothing should have been deleted.
assert _all_urls(seeded_db) == urls_before
def test_bulk_delete_with_confirmed_actually_deletes(seeded_db, csrf_session):
pid = _page_id(seeded_db, "https://example.com/rust-intro")
resp = handle_bulk_action({
"ids": [str(pid)],
"action": ["delete"],
"confirmed": ["1"],
})
# Confirmed delete redirects back to /pages.
assert resp["status"] in (302, 303)
urls = _all_urls(seeded_db)
assert "https://example.com/rust-intro" not in urls
# Other pages untouched.
assert "https://example.com/python-tips" in urls
def test_bulk_delete_with_no_ids_redirects(seeded_db, csrf_session):
resp = handle_bulk_action({
"ids": [],
"action": ["delete"],
"confirmed": ["1"],
})
assert resp["status"] in (302, 303)
assert _all_urls(seeded_db) == {
"https://example.com/rust-intro",
"https://example.com/python-tips",
"https://example.com/ocaml-why",
"https://news.example.org/mesh",
}
def test_bulk_delete_rejects_non_integer_ids(seeded_db, csrf_session):
resp = handle_bulk_action({
"ids": ["not-a-number"],
"action": ["delete"],
"confirmed": ["1"],
})
assert resp["status"] == 400
def test_bulk_retag_add_mode_merges_tags(seeded_db, csrf_session):
pid = _page_id(seeded_db, "https://example.com/python-tips")
handle_bulk_action({
"ids": [str(pid)],
"action": ["retag"],
"bulk_tags": ["scripting, tutorials"],
"tag_mode": ["add"],
})
db = get_db()
try:
rows = db.execute(
"SELECT t.name FROM tags t JOIN page_tags pt ON pt.tag_id = t.id "
"WHERE pt.page_id = ? ORDER BY t.name",
(pid,),
).fetchall()
finally:
return_db(db)
tags = [r["name"] for r in rows]
assert "python" in tags # existing kept
assert "scripting" in tags # new added
assert "tutorials" in tags
def test_bulk_retag_replace_mode_overwrites_tags(seeded_db, csrf_session):
pid = _page_id(seeded_db, "https://example.com/python-tips")
handle_bulk_action({
"ids": [str(pid)],
"action": ["retag"],
"bulk_tags": ["one, two"],
"tag_mode": ["replace"],
})
db = get_db()
try:
rows = db.execute(
"SELECT t.name FROM tags t JOIN page_tags pt ON pt.tag_id = t.id "
"WHERE pt.page_id = ?",
(pid,),
).fetchall()
finally:
return_db(db)
tags = {r["name"] for r in rows}
assert tags == {"one", "two"}
assert "python" not in tags
def test_edit_form_renders_current_values(seeded_db, csrf_session):
pid = _page_id(seeded_db, "https://example.com/rust-intro")
resp = handle_edit_form(pid)
assert resp["status"] == 200
assert "Rust Intro" in resp["body"]
# Existing tags should appear in the tag field.
assert "rust" in resp["body"]
def test_edit_form_404_for_unknown_page(temp_db, csrf_session):
resp = handle_edit_form(99999)
assert resp["status"] == 404
def test_edit_submit_updates_title_and_note(seeded_db, csrf_session):
pid = _page_id(seeded_db, "https://example.com/rust-intro")
handle_edit_submit(pid, {
"title": ["New Rust Title"],
"note": ["new annotation"],
"tags": ["rust, updated"],
})
db = get_db()
try:
row = db.execute("SELECT title, note FROM pages WHERE id = ?", (pid,)).fetchone()
finally:
return_db(db)
assert row["title"] == "New Rust Title"
assert row["note"] == "new annotation"
def test_handle_pages_lists_indexed_pages(seeded_db, csrf_session):
resp = handle_pages({})
assert resp["status"] == 200
# Every seeded page title appears on the list page.
for title in ("Rust Intro", "Python Tips", "Why OCaml", "Mesh Networking"):
assert title in resp["body"]

View file

@ -0,0 +1,63 @@
"""Tests for `handle_search` — the home page + primary user flow."""
from handlers import handle_search
def test_empty_index_empty_query_shows_welcome(temp_db, csrf_session):
resp = handle_search({})
assert resp["status"] == 200
body = resp["body"]
assert "Your index is empty" in body
# Links the welcome panel offers as equal-weight starting points.
assert "/add" in body
assert "/style" in body
assert "/subscriptions" in body
def test_empty_index_with_query_shows_no_results(temp_db, csrf_session):
resp = handle_search({"q": ["rust"]})
assert resp["status"] == 200
assert "No results in your index" in resp["body"]
def test_populated_index_with_matching_query_returns_results(seeded_db, csrf_session):
resp = handle_search({"q": ["rust"]})
assert resp["status"] == 200
assert "Rust Intro" in resp["body"]
# Page count shown in meta line.
assert "4 pages indexed" in resp["body"]
def test_query_only_matches_relevant_pages(seeded_db, csrf_session):
resp = handle_search({"q": ["ocaml"]})
body = resp["body"]
assert "Why OCaml" in body
assert "Python Tips" not in body
assert "Rust Intro" not in body
def test_pagination_query_param_respected(seeded_db, csrf_session):
"""A high page number should still render without crashing."""
resp = handle_search({"q": ["example"], "p": ["99"]})
assert resp["status"] == 200
def test_trusted_sites_fallback_surfaces_when_query_matches_link_label(seeded_db, csrf_session):
"""Links extracted from indexed pages act as a fallback when direct results
are absent or thin; labels are substring-matched case-insensitively."""
resp = handle_search({"q": ["advanced"]})
body = resp["body"]
# The label "advanced rust guide" is on a link extracted from rust-intro.
assert "advanced rust guide" in body
assert "trusted sites" in body
def test_page_count_in_meta_line(seeded_db, csrf_session):
resp = handle_search({})
assert "4 pages indexed" in resp["body"]
def test_csp_and_security_headers_not_in_handler_but_via_dispatch(seeded_db, csrf_session):
"""Handler itself returns no security headers; dispatch_request wraps them.
This test documents the boundary so future refactors don't break assumptions."""
resp = handle_search({})
assert "headers" not in resp or "Content-Security-Policy" not in resp.get("headers", {})

112
tests/test_handlers_subs.py Normal file
View file

@ -0,0 +1,112 @@
"""Tests for subscription handlers.
Subscription add validates the destination hash (32-char hex) locally
before calling `fetch_remote_sites`; browse uses cached remote_pages when
available and falls back to a live fetch otherwise.
"""
from unittest.mock import patch
import handlers as handlers_module
from db import get_db, return_db
from handlers import handle_subscription_add, handle_subscription_browse
VALID_HASH = "a" * 32
def _subscription_count():
db = get_db()
try:
return db.execute("SELECT count(*) FROM subscriptions").fetchone()[0]
finally:
return_db(db)
def test_rejects_empty_dest_hash(temp_db, csrf_session):
resp = handle_subscription_add({"dest_hash": [""]})
assert "32-character" in resp["body"]
assert _subscription_count() == 0
def test_rejects_wrong_length(temp_db, csrf_session):
resp = handle_subscription_add({"dest_hash": ["abc123"]})
assert "32-character" in resp["body"]
assert _subscription_count() == 0
def test_rejects_non_hex(temp_db, csrf_session):
resp = handle_subscription_add({"dest_hash": ["z" * 32]})
assert "hex" in resp["body"].lower()
assert _subscription_count() == 0
def test_rejects_unreachable_peer(temp_db, csrf_session):
with patch.object(handlers_module, "fetch_remote_sites") as fetch:
fetch.side_effect = ConnectionError("unreachable")
resp = handle_subscription_add({"dest_hash": [VALID_HASH]})
assert "Could not reach" in resp["body"]
assert _subscription_count() == 0
def test_rejects_peer_with_sharing_disabled(temp_db, csrf_session):
with patch.object(handlers_module, "fetch_remote_sites") as fetch:
fetch.side_effect = PermissionError("sharing disabled")
resp = handle_subscription_add({"dest_hash": [VALID_HASH]})
assert "sharing disabled" in resp["body"]
assert _subscription_count() == 0
def test_successful_add_records_subscription(temp_db, csrf_session):
with patch.object(handlers_module, "fetch_remote_sites") as fetch:
fetch.return_value = {"name": "alice", "sites": []}
resp = handle_subscription_add({"dest_hash": [VALID_HASH]})
assert "Subscribed to alice" in resp["body"]
assert _subscription_count() == 1
def test_dest_hash_strips_angle_brackets(temp_db, csrf_session):
"""Users often paste hashes as `<aaa...>` from RNS log output; strip them."""
with patch.object(handlers_module, "fetch_remote_sites") as fetch:
fetch.return_value = {"name": "bob", "sites": []}
resp = handle_subscription_add({"dest_hash": [f"<{VALID_HASH}>"]})
assert _subscription_count() == 1
def test_browse_unknown_subscription_is_404(temp_db, csrf_session):
resp = handle_subscription_browse(99999)
assert resp["status"] == 404
def test_browse_marks_already_indexed_urls(seeded_db, csrf_session):
# Insert a subscription + some remote pages (one duplicate of local, one new).
db = get_db()
try:
db.execute(
"INSERT INTO subscriptions (dest_hash, name) VALUES (?, ?)",
(VALID_HASH, "alice"),
)
sub_id = db.execute("SELECT id FROM subscriptions").fetchone()["id"]
db.execute(
"INSERT INTO remote_pages (subscription_id, url, title, note, tags) "
"VALUES (?, ?, ?, ?, ?)",
(sub_id, "https://example.com/rust-intro", "Alice rust pick", "", ""),
)
db.execute(
"INSERT INTO remote_pages (subscription_id, url, title, note, tags) "
"VALUES (?, ?, ?, ?, ?)",
(sub_id, "https://new.example.com/shiny", "Shiny New Link", "note", "tag1"),
)
db.commit()
finally:
return_db(db)
resp = handle_subscription_browse(sub_id)
body = resp["body"]
assert resp["status"] == 200
assert "already indexed" in body
# The duplicate URL should appear in the "already indexed" section.
assert "Alice rust pick" in body
# The new URL should be in the selectable section.
assert "Shiny New Link" in body
# Count summary: "2 site(s) available, 1 new"
assert "1 new" in body

101
tests/test_handlers_tags.py Normal file
View file

@ -0,0 +1,101 @@
"""Tests for tag helpers and the tag browse handler.
Tags are stored via a join table, so orphaned rows in `tags` can accumulate
if `_cleanup_orphaned_tags` isn't called after deletion/retagging. Tag
counts shown in the UI rely on this being right.
"""
from db import get_db, return_db
from handlers import (
_cleanup_orphaned_tags,
_get_page_tags,
_set_page_tags,
handle_tag_browse,
handle_tags,
)
def _page_id(url):
db = get_db()
try:
row = db.execute("SELECT id FROM pages WHERE url = ?", (url,)).fetchone()
return row["id"] if row else None
finally:
return_db(db)
def _tag_names():
db = get_db()
try:
return {r["name"] for r in db.execute("SELECT name FROM tags").fetchall()}
finally:
return_db(db)
def test_get_page_tags_returns_sorted_names(seeded_db):
pid = _page_id("https://example.com/rust-intro")
tags = _get_page_tags(pid)
assert tags == sorted(tags) # alphabetical
assert "rust" in tags
assert "public" in tags
def test_set_page_tags_replaces_existing(seeded_db):
pid = _page_id("https://example.com/rust-intro")
db = get_db()
try:
_set_page_tags(pid, "brand, new, tags", db)
db.commit()
finally:
return_db(db)
current = _get_page_tags(pid)
assert current == ["brand", "new", "tags"]
def test_set_page_tags_splits_on_comma_and_lowercases(seeded_db):
pid = _page_id("https://example.com/python-tips")
db = get_db()
try:
_set_page_tags(pid, "Foo, BAR, baz", db)
db.commit()
finally:
return_db(db)
assert set(_get_page_tags(pid)) == {"foo", "bar", "baz"}
def test_cleanup_orphaned_tags_removes_unreferenced(seeded_db):
# Clear all tags on one page; previously-unique tags become orphans.
pid = _page_id("https://example.com/rust-intro")
db = get_db()
try:
_set_page_tags(pid, "", db) # empty string = no tags
# `rust` was only on the rust-intro page; `public` is also on mesh.
_cleanup_orphaned_tags(db)
db.commit()
finally:
return_db(db)
names = _tag_names()
assert "rust" not in names # pruned
assert "public" in names # still on mesh
def test_handle_tag_browse_filters_by_tag(seeded_db, csrf_session):
resp = handle_tag_browse("rust", {})
assert resp["status"] == 200
body = resp["body"]
assert "Rust Intro" in body
assert "Python Tips" not in body
assert "Why OCaml" not in body
def test_handle_tag_browse_unknown_tag_is_graceful(seeded_db, csrf_session):
resp = handle_tag_browse("no-such-tag", {})
# Should render a valid page with zero results, not error.
assert resp["status"] == 200
def test_handle_tags_lists_all_tags_with_counts(seeded_db, csrf_session):
resp = handle_tags()
assert resp["status"] == 200
body = resp["body"]
for tag in ("rust", "python", "ocaml", "mesh", "public", "private"):
assert tag in body

View file

@ -0,0 +1,138 @@
"""Tests for link extraction inside `fetch_page`.
Link extraction powers the "trusted sites" fallback on empty searches and
feeds the `links` table. Rules: same-domain only, skip binary extensions,
skip Wikipedia special pages, resolve relatives via urljoin.
"""
from unittest.mock import patch
from conftest import patch_dns_ok
import db as db_module
class FakeResponse:
def __init__(self, text, status_code=200):
self.text = text
self.status_code = status_code
self.is_redirect = False
self.headers = {}
def raise_for_status(self):
if self.status_code >= 400:
raise Exception(f"status {self.status_code}")
def _fetch_with_html(monkeypatch, url, html):
"""Invoke fetch_page against `url` with `html` as the mocked response body."""
patch_dns_ok(monkeypatch)
with patch.object(db_module, "requests") as mock_requests:
mock_requests.get.return_value = FakeResponse(html)
return db_module.fetch_page(url)
def test_only_same_domain_links_kept(monkeypatch):
html = """
<html><body>
<a href="https://example.com/a">same</a>
<a href="https://other.com/b">cross</a>
<a href="https://sub.example.com/c">subdomain</a>
</body></html>
"""
_, _, links, _ = _fetch_with_html(monkeypatch, "https://example.com/", html)
urls = [u for u, _label in links]
assert "https://example.com/a" in urls
assert "https://other.com/b" not in urls
assert "https://sub.example.com/c" not in urls
def test_binary_extensions_skipped(monkeypatch):
html = """
<html><body>
<a href="/real-page">keep</a>
<a href="/image.png">skip</a>
<a href="/doc.pdf">skip</a>
<a href="/archive.zip">skip</a>
<a href="/song.mp3">skip</a>
<a href="/styles.css">skip</a>
</body></html>
"""
_, _, links, _ = _fetch_with_html(monkeypatch, "https://example.com/", html)
urls = [u for u, _label in links]
assert "https://example.com/real-page" in urls
for ext in (".png", ".pdf", ".zip", ".mp3", ".css"):
assert not any(u.endswith(ext) for u in urls), f"{ext} leaked through"
def test_wikipedia_special_pages_skipped(monkeypatch):
html = """
<html><body>
<a href="/wiki/Main_Page">keep</a>
<a href="/wiki/Special:Random">skip</a>
<a href="/wiki/Talk:Foo">skip</a>
<a href="/wiki/User:Jimbo">skip</a>
<a href="/wiki/Category:Bar">skip</a>
</body></html>
"""
_, _, links, _ = _fetch_with_html(monkeypatch, "https://example.com/", html)
urls = [u for u, _label in links]
assert "https://example.com/wiki/Main_Page" in urls
for skip in ("Special:Random", "Talk:Foo", "User:Jimbo", "Category:Bar"):
assert not any(skip in u for u in urls), f"wiki {skip!r} leaked"
def test_relative_urls_resolved(monkeypatch):
html = """<html><body><a href="/relative/path">r</a></body></html>"""
_, _, links, _ = _fetch_with_html(monkeypatch, "https://example.com/start", html)
urls = [u for u, _label in links]
assert "https://example.com/relative/path" in urls
def test_fragment_stripped_from_extracted_links(monkeypatch):
html = """<html><body><a href="/page#section">r</a></body></html>"""
_, _, links, _ = _fetch_with_html(monkeypatch, "https://example.com/", html)
urls = [u for u, _label in links]
assert "https://example.com/page" in urls
assert not any("#" in u for u in urls)
def test_duplicate_links_deduped(monkeypatch):
html = """
<html><body>
<a href="/a">first</a>
<a href="/a">second</a>
<a href="/a">third</a>
</body></html>
"""
_, _, links, _ = _fetch_with_html(monkeypatch, "https://example.com/", html)
urls = [u for u, _label in links]
assert urls.count("https://example.com/a") == 1
def test_label_truncated_to_200(monkeypatch):
long_text = "x" * 500
html = f'<html><body><a href="/p">{long_text}</a></body></html>'
_, _, links, _ = _fetch_with_html(monkeypatch, "https://example.com/", html)
assert len(links) == 1
_, label = links[0]
assert len(label) <= 200
def test_meta_description_extracted(monkeypatch):
html = """
<html><head>
<meta name="description" content="the real description">
</head><body><p>body content</p></body></html>
"""
title, body, links, meta = _fetch_with_html(monkeypatch, "https://example.com/", html)
assert meta == "the real description"
def test_og_description_fallback(monkeypatch):
"""When there's no <meta name=description>, og:description wins."""
html = """
<html><head>
<meta property="og:description" content="open graph fallback">
</head><body><p>body</p></body></html>
"""
_, _, _, meta = _fetch_with_html(monkeypatch, "https://example.com/", html)
assert meta == "open graph fallback"

58
tests/test_pagination.py Normal file
View file

@ -0,0 +1,58 @@
"""Tests for `_paginate` and `_page_nav`."""
from handlers import _paginate, _page_nav, PER_PAGE
def test_paginate_default_is_one():
assert _paginate({}) == 1
def test_paginate_reads_query_string():
assert _paginate({"p": ["3"]}) == 3
def test_paginate_clamps_to_one():
assert _paginate({"p": ["0"]}) == 1
assert _paginate({"p": ["-5"]}) == 1
def test_paginate_handles_bad_input():
assert _paginate({"p": ["not-a-number"]}) == 1
assert _paginate({"p": []}) == 1
def test_paginate_custom_key():
assert _paginate({"batch": ["7"]}, key="batch") == 7
def test_page_nav_empty_when_single_page():
assert _page_nav(1, PER_PAGE, "/?q=foo") == ""
assert _page_nav(1, 0, "/?q=foo") == ""
def test_page_nav_shows_next_on_first_page():
out = _page_nav(1, PER_PAGE * 3, "/?q=foo")
assert "next" in out
assert "prev" not in out
assert "page 1 of 3" in out
def test_page_nav_shows_both_in_middle():
out = _page_nav(2, PER_PAGE * 3, "/?q=foo")
assert "next" in out
assert "prev" in out
def test_page_nav_shows_prev_on_last_page():
out = _page_nav(3, PER_PAGE * 3, "/?q=foo")
assert "next" not in out
assert "prev" in out
assert "page 3 of 3" in out
def test_page_nav_handles_query_string_separator():
# when base_url already has ?, pagination links must use &
out = _page_nav(1, PER_PAGE * 2, "/?q=foo")
assert "&p=2" in out
# when base_url has no ?, pagination links use ?
out = _page_nav(1, PER_PAGE * 2, "/pages")
assert "?p=2" in out

107
tests/test_regressions.py Normal file
View file

@ -0,0 +1,107 @@
"""Aggregator of regression tests tied to specific bug-fix commits.
Each test here guards against a specific bug that was once shipped. Running
just this file gives a one-line-per-bug audit:
pytest tests/test_regressions.py -v
The test bodies are intentionally small; for the exhaustive behavior of each
module, see the topical test files (test_fts_sanitizer.py, test_url_cleanup.py,
etc.). This file's job is to make the bug catalog scannable.
"""
import socket
from unittest.mock import patch
import pytest
import app as app_module
import db as db_module
import handlers as handlers_module
from conftest import patch_dns_fail, patch_dns_ok
from db import clean_url
from handlers import _sanitize_fts_query, handle_bulk_action
def test_6ffd38d_clean_url_preserves_www_when_bare_domain_fails(monkeypatch):
"""6ffd38d: `clean_url` used to strip `www.` unconditionally; for sites that
only serve at `www.`, this produced unreachable clean URLs."""
patch_dns_fail(monkeypatch)
assert clean_url("https://www.example.com/page") == "https://www.example.com/page"
def test_1bc695f_fts_sanitizer_strips_colon():
"""1bc695f: FTS5 colon is a column filter — must not appear in sanitized output."""
assert ":" not in _sanitize_fts_query("title:secret body:exposed")
@pytest.mark.parametrize("op", ["AND", "OR", "NOT", "NEAR"])
def test_1bc695f_fts_sanitizer_drops_operator_words(op):
"""1bc695f: operator words (AND/OR/NOT/NEAR) would be interpreted as FTS5
operators if they landed on the unquoted last token."""
out = _sanitize_fts_query(f"foo {op} bar")
# operator itself should not appear in the output
tokens = out.replace('"', '').split()
assert op not in [t.rstrip("*") for t in tokens]
def test_1bc695f_gateway_rejects_oversize_body():
"""1bc695f: 16 MiB body-size cap prevents memory-exhaustion DoS."""
from tests.test_gateway_limits import FakeGatewayHandler
from gateway import MAX_BODY_SIZE
h = FakeGatewayHandler(
path="/add", method="POST",
headers={"Content-Length": str(MAX_BODY_SIZE + 1)},
)
h._forward("POST")
assert h._captured["error"] and h._captured["error"][0] == 413
def test_1bc695f_mesh_rejects_non_whitelisted_paths():
"""1bc695f: Reticulum callers are limited to GET /api/sites; CSRF cannot
authenticate mesh callers."""
resp = app_module.rns_request_handler(
path="/tinyweb",
data={"method": "POST", "path": "/add", "query": {}, "body": {}, "gateway_host": ""},
request_id="x", link_id="y", remote_identity=None, requested_at=0,
)
assert resp["status"] == 403
def test_1bc695f_pool_returns_clean_connection(temp_db, monkeypatch):
"""1bc695f: uncommitted transactions on a pooled connection used to leak
into the next consumer."""
from db import get_db, return_db
db = get_db()
db.execute(
"INSERT INTO pages (url, title, body) VALUES (?, ?, ?)",
("https://leak.example.com/", "should not persist", "body"),
)
return_db(db) # no commit
db2 = get_db()
try:
urls = {r["url"] for r in db2.execute("SELECT url FROM pages").fetchall()}
finally:
return_db(db2)
assert "https://leak.example.com/" not in urls
def test_8dffd8c_bulk_delete_requires_confirmation(seeded_db, csrf_session):
"""8dffd8c: bulk delete without confirmed=1 must render a confirm page
instead of deleting the JS confirm on /pages is a first-line filter only."""
from db import get_db, return_db
db = get_db()
try:
pid = db.execute("SELECT id FROM pages LIMIT 1").fetchone()["id"]
count_before = db.execute("SELECT count(*) FROM pages").fetchone()[0]
finally:
return_db(db)
resp = handle_bulk_action({"ids": [str(pid)], "action": ["delete"]})
assert "confirm delete" in resp["body"].lower()
db = get_db()
try:
count_after = db.execute("SELECT count(*) FROM pages").fetchone()[0]
finally:
return_db(db)
assert count_before == count_after, "bulk delete ran without confirmation"

View file

@ -0,0 +1,38 @@
"""Tests for `_page_is_shared`.
This function decides whether a page is exposed over Reticulum to
subscribers. Getting it wrong means either a privacy leak or silently
hiding pages the user meant to share both are worth a regression net.
"""
import pytest
from handlers import _page_is_shared
@pytest.mark.parametrize("mode", ["exclude_private", "require_public"])
def test_private_tag_always_excludes(mode):
"""`private` tag overrides every mode — the most important invariant."""
assert _page_is_shared(["private"], mode) is False
assert _page_is_shared(["public", "private"], mode) is False
def test_exclude_private_defaults_to_shared():
assert _page_is_shared([], "exclude_private") is True
assert _page_is_shared(["random-tag"], "exclude_private") is True
def test_require_public_needs_public_tag():
assert _page_is_shared([], "require_public") is False
assert _page_is_shared(["rust"], "require_public") is False
assert _page_is_shared(["public"], "require_public") is True
def test_require_public_still_vetoes_private():
# public AND private → private wins.
assert _page_is_shared(["public", "private"], "require_public") is False
def test_unknown_mode_treated_as_exclude_private():
"""The default mode is 'exclude_private'; unknown modes fall through to it."""
assert _page_is_shared([], "totally-bogus-mode") is True
assert _page_is_shared(["private"], "totally-bogus-mode") is False

64
tests/test_ssrf.py Normal file
View file

@ -0,0 +1,64 @@
"""Tests for `_validate_url_target` — SSRF prevention.
Any URL the app fetches must resolve to a public IP; private/internal/
loopback addresses must be rejected so attacker-controlled URLs cannot
reach internal services via our HTTP client.
"""
import socket
from unittest.mock import patch
import pytest
from db import _validate_url_target
def _mock_getaddrinfo(address):
"""Return a function suitable as a socket.getaddrinfo replacement."""
def f(host, port, *args, **kwargs):
family = socket.AF_INET6 if ":" in address else socket.AF_INET
return [(family, socket.SOCK_STREAM, 0, "", (address, port or 80))]
return f
@pytest.mark.parametrize("blocked_ip", [
"127.0.0.1",
"127.1.2.3",
"10.0.0.1",
"10.255.255.255",
"172.16.0.1",
"172.31.255.255",
"192.168.0.1",
"192.168.255.255",
"169.254.169.254",
"0.0.0.0",
"::1",
"fc00::1",
"fe80::1",
])
def test_blocks_private_and_loopback(monkeypatch, blocked_ip):
monkeypatch.setattr(socket, "getaddrinfo", _mock_getaddrinfo(blocked_ip))
with pytest.raises(ValueError, match="blocked"):
_validate_url_target("https://evil.example.com/internal")
def test_allows_public_ipv4(monkeypatch):
monkeypatch.setattr(socket, "getaddrinfo", _mock_getaddrinfo("8.8.8.8"))
_validate_url_target("https://dns.example.com/") # does not raise
def test_allows_public_ipv6(monkeypatch):
monkeypatch.setattr(socket, "getaddrinfo", _mock_getaddrinfo("2001:4860:4860::8888"))
_validate_url_target("https://v6.example.com/") # does not raise
def test_rejects_unresolvable_hostname(monkeypatch):
def boom(*args, **kwargs):
raise socket.gaierror("no such host")
monkeypatch.setattr(socket, "getaddrinfo", boom)
with pytest.raises(ValueError, match="Cannot resolve"):
_validate_url_target("https://does-not-exist.example.com/")
def test_rejects_missing_hostname():
with pytest.raises(ValueError, match="No hostname"):
_validate_url_target("http:///path-only")

101
tests/test_url_cleanup.py Normal file
View file

@ -0,0 +1,101 @@
"""Tests for `clean_url` — URL normalization and tracking-param stripping.
Clean URLs are the deduplication key in the pages table, so any change to
this function can silently cause duplicate rows or mask legitimate saves.
"""
import pytest
from conftest import patch_dns_ok, patch_dns_fail
from db import clean_url, TRACKING_PARAMS
def test_strips_fragment(monkeypatch):
patch_dns_ok(monkeypatch)
assert clean_url("https://example.com/page#section") == "https://example.com/page"
def test_prefers_https(monkeypatch):
patch_dns_ok(monkeypatch)
assert clean_url("http://example.com/page") == "https://example.com/page"
def test_lowercases_hostname(monkeypatch):
patch_dns_ok(monkeypatch)
assert clean_url("https://EXAMPLE.COM/page") == "https://example.com/page"
def test_preserves_path_case(monkeypatch):
"""Paths are case-sensitive and should not be lowercased."""
patch_dns_ok(monkeypatch)
assert clean_url("https://example.com/Foo/Bar") == "https://example.com/Foo/Bar"
def test_strips_default_https_port(monkeypatch):
patch_dns_ok(monkeypatch)
assert clean_url("https://example.com:443/page") == "https://example.com/page"
@pytest.mark.xfail(reason="clean_url upgrades http->https before the port-default check, "
"so port 80 is not stripped. Minor dedup bug — harmless but worth fixing.")
def test_strips_http_port_80(monkeypatch):
"""Expected: http://foo:80 → https://foo (both scheme-upgrade and port-strip).
Currently fails because scheme is upgraded to https *before* the port check,
so `scheme == "http" and port == 80` is never true by the time the check runs.
"""
patch_dns_ok(monkeypatch)
assert clean_url("http://example.com:80/page") == "https://example.com/page"
def test_preserves_non_default_port(monkeypatch):
patch_dns_ok(monkeypatch)
assert clean_url("https://example.com:8443/page") == "https://example.com:8443/page"
def test_strips_trailing_slash(monkeypatch):
patch_dns_ok(monkeypatch)
assert clean_url("https://example.com/page/") == "https://example.com/page"
def test_root_slash_preserved(monkeypatch):
patch_dns_ok(monkeypatch)
assert clean_url("https://example.com/") == "https://example.com/"
@pytest.mark.parametrize("param", sorted(TRACKING_PARAMS))
def test_tracking_params_stripped(monkeypatch, param):
patch_dns_ok(monkeypatch)
result = clean_url(f"https://example.com/page?{param}=value&keep=yes")
assert param not in result
assert "keep=yes" in result
def test_strips_www_when_nonwww_resolves(monkeypatch):
"""Standard case: strip `www.` prefix to canonicalize."""
patch_dns_ok(monkeypatch)
assert clean_url("https://www.example.com/page") == "https://example.com/page"
def test_preserves_www_when_nonwww_does_not_resolve(monkeypatch):
"""Regression for 6ffd38d.
Some sites only serve their content at `www.domain.tld`; the bare domain
doesn't resolve. Stripping `www.` in that case produced a URL that we could
never actually fetch or dedupe against the real one.
"""
patch_dns_fail(monkeypatch)
assert clean_url("https://www.example.com/page") == "https://www.example.com/page"
def test_query_params_sorted_for_stable_ordering(monkeypatch):
"""Same URL with different param orderings should produce the same clean URL."""
patch_dns_ok(monkeypatch)
a = clean_url("https://example.com/page?b=2&a=1")
b = clean_url("https://example.com/page?a=1&b=2")
assert a == b
def test_path_and_query_preserved_through_cleanup(monkeypatch):
patch_dns_ok(monkeypatch)
result = clean_url("https://example.com/path/to/page?id=42&utm_source=twitter")
assert result == "https://example.com/path/to/page?id=42"

View file

@ -3,15 +3,16 @@
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<meta name="referrer" content="no-referrer">
<meta http-equiv="x-dns-prefetch-control" content="off">
<style>
@import url('https://fonts.googleapis.com/css2?family=Nunito:ital,wght@0,400;0,600;0,700;1,400&family=Fira+Code:wght@400;500&display=swap');
* { margin: 0; padding: 0; box-sizing: border-box; }
html, body { min-height: 100vh; }
body {
font-family: 'Nunito', -apple-system, sans-serif;
font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', 'Noto Sans', Helvetica, Arial, sans-serif;
font-size: 16px;
line-height: 1.65;
color: #e0d8c8;
@ -81,7 +82,7 @@
right: 10px;
z-index: 900;
pointer-events: none;
font-family: 'Fira Code', monospace;
font-family: ui-monospace, 'SF Mono', 'Cascadia Code', 'Segoe UI Mono', Menlo, Consolas, monospace;
display: flex;
flex-direction: column;
align-items: flex-end;
@ -289,7 +290,7 @@
}
nav .site {
font-family: 'Fira Code', monospace;
font-family: ui-monospace, 'SF Mono', 'Cascadia Code', 'Segoe UI Mono', Menlo, Consolas, monospace;
font-size: 0.85rem;
font-weight: 500;
color: #f0d878;
@ -383,7 +384,7 @@
border-radius: 4px;
padding: 0.6rem 0.85rem;
color: #d0c8b8;
font-family: 'Nunito', sans-serif;
font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', 'Noto Sans', Helvetica, Arial, sans-serif;
font-size: 0.95rem;
transition: border-color 0.2s, box-shadow 0.3s;
}
@ -400,7 +401,7 @@
border-radius: 4px;
padding: 0.6rem 1.1rem;
color: #a098b0;
font-family: 'Nunito', sans-serif;
font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', 'Noto Sans', Helvetica, Arial, sans-serif;
font-size: 0.88rem;
transition: all 0.2s;
}
@ -443,7 +444,7 @@
.tags { margin-top: 0.3rem; }
.tag, .tags a {
font-family: 'Fira Code', monospace;
font-family: ui-monospace, 'SF Mono', 'Cascadia Code', 'Segoe UI Mono', Menlo, Consolas, monospace;
font-size: 0.7rem;
color: #8878a0;
border: 1px solid rgba(255,255,255,0.08);
@ -486,7 +487,7 @@
/* code */
pre {
font-family: 'Fira Code', monospace;
font-family: ui-monospace, 'SF Mono', 'Cascadia Code', 'Segoe UI Mono', Menlo, Consolas, monospace;
font-size: 0.8rem;
background: rgba(15, 10, 40, 0.5);
border: 1px solid rgba(255,255,255,0.08);
@ -498,7 +499,7 @@
}
code {
font-family: 'Fira Code', monospace;
font-family: ui-monospace, 'SF Mono', 'Cascadia Code', 'Segoe UI Mono', Menlo, Consolas, monospace;
font-size: 0.82rem;
background: rgba(15, 10, 40, 0.4);
border-radius: 3px;
@ -513,7 +514,7 @@
border-radius: 4px;
padding: 0.7rem 0.9rem;
color: #d0c8b8;
font-family: 'Fira Code', monospace;
font-family: ui-monospace, 'SF Mono', 'Cascadia Code', 'Segoe UI Mono', Menlo, Consolas, monospace;
font-size: 0.8rem;
line-height: 1.6;
resize: vertical;
@ -547,7 +548,7 @@
hr { border: none; border-top: 1px solid rgba(255,255,255,0.06); margin: 1rem 0; }
small {
font-family: 'Fira Code', monospace;
font-family: ui-monospace, 'SF Mono', 'Cascadia Code', 'Segoe UI Mono', Menlo, Consolas, monospace;
font-size: 0.7rem;
color: #5a5070;
}
@ -562,7 +563,7 @@
}
footer .clock {
font-family: 'Fira Code', monospace;
font-family: ui-monospace, 'SF Mono', 'Cascadia Code', 'Segoe UI Mono', Menlo, Consolas, monospace;
font-size: 0.72rem;
color: #3a3050;
margin-top: 0.25rem;

View file

@ -3,13 +3,14 @@
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<meta name="referrer" content="no-referrer">
<meta http-equiv="x-dns-prefetch-control" content="off">
<style>
@import url('https://fonts.googleapis.com/css2?family=IBM+Plex+Mono:wght@400;500&family=IBM+Plex+Sans:ital,wght@0,400;0,500;0,600;1,400&display=swap');
* { margin: 0; padding: 0; box-sizing: border-box; }
body {
font-family: 'IBM Plex Sans', -apple-system, sans-serif;
font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', 'Noto Sans', Helvetica, Arial, sans-serif;
font-size: 16px;
line-height: 1.65;
color: #c8c8c8;
@ -99,7 +100,7 @@
}
nav .site {
font-family: 'IBM Plex Mono', monospace;
font-family: ui-monospace, 'SF Mono', 'Cascadia Code', 'Segoe UI Mono', Menlo, Consolas, monospace;
font-size: 0.85rem;
font-weight: 500;
color: #e8e8e8;
@ -193,7 +194,7 @@
border-radius: 4px;
padding: 0.6rem 0.85rem;
color: #d0d0d0;
font-family: 'IBM Plex Sans', sans-serif;
font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', 'Noto Sans', Helvetica, Arial, sans-serif;
font-size: 0.95rem;
transition: border-color 0.2s, box-shadow 0.3s;
}
@ -210,7 +211,7 @@
border-radius: 4px;
padding: 0.6rem 1.1rem;
color: #999;
font-family: 'IBM Plex Sans', sans-serif;
font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', 'Noto Sans', Helvetica, Arial, sans-serif;
font-size: 0.88rem;
transition: all 0.2s;
}
@ -253,7 +254,7 @@
.tags { margin-top: 0.3rem; }
.tag, .tags a {
font-family: 'IBM Plex Mono', monospace;
font-family: ui-monospace, 'SF Mono', 'Cascadia Code', 'Segoe UI Mono', Menlo, Consolas, monospace;
font-size: 0.7rem;
color: #555;
border: 1px solid #252525;
@ -296,7 +297,7 @@
/* code */
pre {
font-family: 'IBM Plex Mono', monospace;
font-family: ui-monospace, 'SF Mono', 'Cascadia Code', 'Segoe UI Mono', Menlo, Consolas, monospace;
font-size: 0.8rem;
background: #151515;
border: 1px solid #232323;
@ -308,7 +309,7 @@
}
code {
font-family: 'IBM Plex Mono', monospace;
font-family: ui-monospace, 'SF Mono', 'Cascadia Code', 'Segoe UI Mono', Menlo, Consolas, monospace;
font-size: 0.82rem;
background: #1a1a1a;
border-radius: 3px;
@ -323,7 +324,7 @@
border-radius: 4px;
padding: 0.7rem 0.9rem;
color: #c8c8c8;
font-family: 'IBM Plex Mono', monospace;
font-family: ui-monospace, 'SF Mono', 'Cascadia Code', 'Segoe UI Mono', Menlo, Consolas, monospace;
font-size: 0.8rem;
line-height: 1.6;
resize: vertical;
@ -357,7 +358,7 @@
hr { border: none; border-top: 1px solid #1e1e1e; margin: 1rem 0; }
small {
font-family: 'IBM Plex Mono', monospace;
font-family: ui-monospace, 'SF Mono', 'Cascadia Code', 'Segoe UI Mono', Menlo, Consolas, monospace;
font-size: 0.7rem;
color: #484848;
}
@ -372,7 +373,7 @@
}
footer .clock {
font-family: 'IBM Plex Mono', monospace;
font-family: ui-monospace, 'SF Mono', 'Cascadia Code', 'Segoe UI Mono', Menlo, Consolas, monospace;
font-size: 0.72rem;
color: #282828;
margin-top: 0.25rem;

View file

@ -3,8 +3,9 @@
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<meta name="referrer" content="no-referrer">
<meta http-equiv="x-dns-prefetch-control" content="off">
<style>
@import url('https://fonts.googleapis.com/css2?family=IBM+Plex+Mono:wght@400;500&family=IBM+Plex+Sans:ital,wght@0,400;0,500;0,600;1,400&display=swap');
* { margin: 0; padding: 0; box-sizing: border-box;
cursor: url("data:image/svg+xml,%3Csvg xmlns='http://www.w3.org/2000/svg' width='20' height='20'%3E%3Ccircle cx='10' cy='10' r='3' fill='none' stroke='%23557766' stroke-width='1.5'/%3E%3Ccircle cx='10' cy='10' r='1' fill='%2377aa88'/%3E%3C/svg%3E") 10 10, default;
@ -13,7 +14,7 @@
html, body { min-height: 100vh; }
body {
font-family: 'IBM Plex Sans', -apple-system, sans-serif;
font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', 'Noto Sans', Helvetica, Arial, sans-serif;
font-size: 16px;
line-height: 1.65;
color: #9ab4b8;
@ -65,7 +66,7 @@
}
nav .site {
font-family: 'IBM Plex Mono', monospace;
font-family: ui-monospace, 'SF Mono', 'Cascadia Code', 'Segoe UI Mono', Menlo, Consolas, monospace;
font-size: 0.85rem;
font-weight: 500;
color: #b0ccc4;
@ -158,7 +159,7 @@
border-radius: 4px;
padding: 0.6rem 0.85rem;
color: #90b4ac;
font-family: 'IBM Plex Sans', sans-serif;
font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', 'Noto Sans', Helvetica, Arial, sans-serif;
font-size: 0.95rem;
transition: border-color 0.2s, box-shadow 0.3s;
}
@ -175,7 +176,7 @@
border-radius: 4px;
padding: 0.6rem 1.1rem;
color: #5a7880;
font-family: 'IBM Plex Sans', sans-serif;
font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', 'Noto Sans', Helvetica, Arial, sans-serif;
font-size: 0.88rem;
transition: all 0.2s;
}
@ -240,7 +241,7 @@
.tags { margin-top: 0.3rem; }
.tag, .tags a {
font-family: 'IBM Plex Mono', monospace;
font-family: ui-monospace, 'SF Mono', 'Cascadia Code', 'Segoe UI Mono', Menlo, Consolas, monospace;
font-size: 0.7rem;
color: #3a5e55;
border: 1px solid rgba(40, 70, 60, 0.35);
@ -273,7 +274,7 @@
li a:hover { border-bottom: 1px solid rgba(80, 130, 110, 0.4); }
pre {
font-family: 'IBM Plex Mono', monospace;
font-family: ui-monospace, 'SF Mono', 'Cascadia Code', 'Segoe UI Mono', Menlo, Consolas, monospace;
font-size: 0.8rem;
background: rgba(6, 14, 16, 0.6);
border: 1px solid rgba(30, 55, 50, 0.3);
@ -285,7 +286,7 @@
}
code {
font-family: 'IBM Plex Mono', monospace;
font-family: ui-monospace, 'SF Mono', 'Cascadia Code', 'Segoe UI Mono', Menlo, Consolas, monospace;
font-size: 0.82rem;
background: rgba(8, 18, 22, 0.6);
border-radius: 3px;
@ -299,7 +300,7 @@
border-radius: 4px;
padding: 0.7rem 0.9rem;
color: #9ab4b8;
font-family: 'IBM Plex Mono', monospace;
font-family: ui-monospace, 'SF Mono', 'Cascadia Code', 'Segoe UI Mono', Menlo, Consolas, monospace;
font-size: 0.8rem;
line-height: 1.6;
resize: vertical;
@ -330,7 +331,7 @@
hr { border: none; border-top: 1px solid rgba(30, 55, 50, 0.3); margin: 1rem 0; }
small {
font-family: 'IBM Plex Mono', monospace;
font-family: ui-monospace, 'SF Mono', 'Cascadia Code', 'Segoe UI Mono', Menlo, Consolas, monospace;
font-size: 0.7rem;
color: #28454e;
}
@ -344,7 +345,7 @@
}
footer .clock {
font-family: 'IBM Plex Mono', monospace;
font-family: ui-monospace, 'SF Mono', 'Cascadia Code', 'Segoe UI Mono', Menlo, Consolas, monospace;
font-size: 0.72rem;
color: #162a30;
margin-top: 0.25rem;