diff --git a/.dockerignore b/.dockerignore
index 031c6d8..9adf733 100644
--- a/.dockerignore
+++ b/.dockerignore
@@ -1,5 +1,15 @@
__pycache__/
+**/__pycache__/
+*.pyc
index.db*
+index.hnsw
tinyweb_identity
.git/
+.gitignore
*.md
+.env
+.env.*
+.venv/
+venv/
+models/
+.DS_Store
diff --git a/.forgejo/workflows/build.yml b/.forgejo/workflows/build.yml
index c7c6fa3..09dd2d6 100644
--- a/.forgejo/workflows/build.yml
+++ b/.forgejo/workflows/build.yml
@@ -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
- if: startsWith(github.ref, 'refs/tags/v')
+ - 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
diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml
deleted file mode 100644
index d79edb3..0000000
--- a/.github/workflows/build.yml
+++ /dev/null
@@ -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 }}
diff --git a/Dockerfile b/Dockerfile
index 9895f65..713df67 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -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
diff --git a/README.md b/README.md
index 371f15d..e44c450 100644
--- a/README.md
+++ b/README.md
@@ -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 --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
diff --git a/app.py b/app.py
index 8302d91..b1c4fe6 100644
--- a/app.py
+++ b/app.py
@@ -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:
diff --git a/conftest.py b/conftest.py
new file mode 100644
index 0000000..9a2f26e
--- /dev/null
+++ b/conftest.py
@@ -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)
diff --git a/db.py b/db.py
index 295da86..8a645ab 100644
--- a/db.py
+++ b/db.py
@@ -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)
diff --git a/embeddings.py b/embeddings.py
index 302a31f..03f6f13 100644
--- a/embeddings.py
+++ b/embeddings.py
@@ -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 embeddings.astype(np.float32)
+ 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()
+ total = db.execute("SELECT COUNT(*) FROM chunks").fetchone()[0]
+ if total == 0:
+ with _hnsw_lock:
+ _hnsw_index = None
+ _hnsw_ids = []
+ return
+
+ 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)
- with _hnsw_lock:
- if not rows:
+ if not all_ids:
+ with _hnsw_lock:
_hnsw_index = None
_hnsw_ids = []
- return
+ 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)
+ 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 = hnswlib.Index(space="cosine", dim=DIMS)
+ index.init_index(max_elements=max(n, 1024), ef_construction=200, M=16)
+ index.add_items(matrix, list(range(n)))
+ 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(
diff --git a/entrypoint.sh b/entrypoint.sh
index e4a9719..1f49fcb 100755
--- a/entrypoint.sh
+++ b/entrypoint.sh
@@ -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 "$@"
diff --git a/gateway.py b/gateway.py
index a13816f..d07924d 100644
--- a/gateway.py
+++ b/gateway.py
@@ -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":
- length = int(self.headers.get("Content-Length", 0))
- raw = self.rfile.read(length).decode()
+ try:
+ length = int(self.headers.get("Content-Length", 0))
+ 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():
diff --git a/handlers.py b/handlers.py
index ae7f484..e47520b 100644
--- a/handlers.py
+++ b/handlers.py
@@ -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''
- f'
{esc(r["title"])} '
+ f'
{esc(r["title"])} '
f'
{esc(r["url"])} '
f'{snip_html}'
f'{note_html}{tags_html}'
@@ -276,7 +285,7 @@ def handle_search(query):
items = ""
for l in trusted:
items += (
- f'
{esc(l["label"])} '
+ f'{esc(l["label"])} '
f'— from {esc(l["source_title"])} '
)
trusted_html = (
@@ -311,8 +320,8 @@ def handle_search(query):
for r in items:
note_html = f' —
{esc(r["note"])} ' if r["note"] else ""
source_items += (
- f'
{esc(r["title"])} '
- f'{note_html} ({esc(r["url"])}) '
+ f'
{esc(r["title"])} '
+ f'{note_html} ({esc(clean_url(r["url"]))}) '
)
remote_html += (
f'
'
@@ -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 = (
+ ''
+ )
return _respond(
f''
f'{count} pages indexed'
f' · + add url
'
+ 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'tag: private to exclude from sharing index '
f""
f"{msg}
"
@@ -421,7 +444,7 @@ def handle_add_submit(body):
f'Title: Description: save manually '
f""
f'back '
@@ -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: {esc(manual_title)} ')
+ return handle_add_form(f'Added manually: {esc(manual_title)} ')
finally:
return_db(db)
@@ -499,8 +522,9 @@ def handle_pages(query=None):
tag_links = " ".join(f'[{esc(t)}] ' for t in tags)
tags_html = f' {tag_links}'
items += (
- f'{esc(r["title"])}{note_html}{tags_html} '
- f'({esc(r["url"])} ) '
+ f' ({esc(r["url"])} ) '
f'edit '
f'remove indexed pages ({total}) "
f"{msg_html}"
+ f''
+ f''
f'export | import
'
f'back '
)
+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'{esc(r["title"] or r["url"])} {esc(r["url"])} confirm delete "
+ f"Remove the following {n} page{'' if n == 1 else 's'}?
"
+ f""
+ f'"
+ f' cancel '
+ )
+
+
+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'Note (why you saved this): Tags (comma-separated): (tag: private to keep private) save '
f""
f"{msg}
"
@@ -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"customize "
f"name your search engine "
@@ -694,13 +848,57 @@ def handle_style_form(msg=""):
f'sharing "
f''
+ f"What to share: privatepublicThe private tag always excludes a page, even in public-only mode. '
+ f'
'
+ f''
+ f'Currently sharing {shared_count} page(s). '
+ f'preview what subscribers would see '
+ f'
'
f"mesh network "
- f"Connect to a Reticulum transport node to reach other peers.
"
+ f"Choose how to connect to the mesh. You can enable both for maximum reach.
"
+ f"internet "
+ f'Reach peers anywhere online. LoRa "
+ f'Reach nearby peers off-grid with an RNode . '
+ f'Serial port: advanced radio settings '
+ f'
search "
f"ai "
f'Uses a 22MB model. Adds ~50ms per search. Disable for faster results. Saves ~50% on storage for embeddings. Slight quality reduction at large scale. manage semantic index custom html "
@@ -724,10 +925,16 @@ def handle_style_form(msg=""):
f"Drag this link to your bookmarks bar. Click it on any page to index it instantly.
"
f'+ save to {esc(name)}
'
f"reset "
- f'"
+ f"maintenance "
+ f'"
f"{msg}
"
f'back ',
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''
f'{sharing_html}'
f'{hash_html}'
+ f'your data '
+ f'Everything is stored locally under ~/.tinyweb/:
'
+ f''
+ f'tinyweb_identity — 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.index.db — your full reading history: every page, '
+ f'note, tag, and synced remote page.models/ — the semantic search model if you enabled it '
+ f'(redownloadable, safe to delete). '
+ f'Back up ~/.tinyweb/ periodically. '
+ f'Copying the whole directory to another device preserves your identity and index together. '
+ f'The export 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.
'
f'what is the slow web? '
f'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'[{esc(t)}] ' for t in tags)
items += (
f'
{esc(r["title"])}{note_html} {tag_links} '
- f'({esc(r["url"])} ) '
+ f'({esc(r["url"])} ) '
)
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 public"
+ if mode == "require_public"
+ else "all pages except those tagged private"
+ )
+ sharing_on = get_setting("sharing_enabled", "0") == "1"
+ status = (
+ 'Sharing is enabled . Subscribers see the pages listed below.
'
+ if sharing_on else
+ 'Sharing is disabled . Nothing is actually being shared right now; '
+ 'this is the list that would be exposed if you enabled it.
'
+ )
+ db = get_db()
+ try:
+ sites = _shared_sites(db)
+ finally:
+ return_db(db)
+ if not sites:
+ body = (
+ "sharing preview "
+ f"Rule: {mode_label}.
"
+ f"{status}"
+ "No pages match the current rule.
"
+ 'back to settings
'
+ )
+ 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' — {esc(s["note"])} ' if s["note"] else ""
+ rows += (
+ f''
+ f'{esc(s["title"] or s["url"])} '
+ f'{note_html}{tags_html} '
+ f'{esc(s["url"])} '
+ f' '
+ )
+ body = (
+ "sharing preview "
+ f"Rule: {mode_label}.
"
+ f"{status}"
+ f"{len(sites)} page(s) visible to subscribers.
"
+ f""
+ 'back to settings
'
+ )
+ 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 = 'syncing...
'
+ elif sync_status.startswith("error:"):
+ err_msg = sync_status[6:]
+ status_html = f'{esc(err_msg)}
'
+ else:
+ status_html = ""
+
+ # Disable sync button while syncing
+ if is_syncing:
+ sync_btn = 'syncing... '
+ else:
+ sync_btn = (
+ f''
+ )
+
cards += (
f''
f'
{esc(s["name"] or "unknown")}
'
f'
{esc(s["dest_hash"])}
'
f'
last sync: {esc(last)}
'
+ f'{status_html}'
f'
'
- f'
browse '
- f'
'
- f'
'
- f'
'
f'
'
f'
syncing... ' if any_syncing else 'sync all '
listing = (
f'{cards}'
f''
+ f'{_csrf_field()}{syncall_btn}'
)
return _respond(
f"subscriptions "
@@ -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"semantic search index "
f"Semantic search is disabled. Enable it in settings to use embeddings.
"
@@ -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("403 Forbidden Invalid or missing CSRF token.
", 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'; "
diff --git a/pytest.ini b/pytest.ini
new file mode 100644
index 0000000..16d6cc5
--- /dev/null
+++ b/pytest.ini
@@ -0,0 +1,5 @@
+[pytest]
+testpaths = tests
+python_files = test_*.py
+filterwarnings =
+ ignore::DeprecationWarning
diff --git a/requirements-dev.txt b/requirements-dev.txt
new file mode 100644
index 0000000..26b77f6
--- /dev/null
+++ b/requirements-dev.txt
@@ -0,0 +1,2 @@
+-r requirements.txt
+pytest
diff --git a/rns_client.py b/rns_client.py
index dbc0af5..98df406 100644
--- a/rns_client.py
+++ b/rns_client.py
@@ -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()
diff --git a/templates.py b/templates.py
index 48beace..0dd9975 100644
--- a/templates.py
+++ b/templates.py
@@ -7,13 +7,13 @@ def esc(s):
-DEFAULT_TEMPLATE = "\n\n\n\n{{content}}\n\n"
+DEFAULT_TEMPLATE = "\n\n{name} search | browse '
' | tags | subscriptions '
diff --git a/tests/test_csrf.py b/tests/test_csrf.py
new file mode 100644
index 0000000..43b4487
--- /dev/null
+++ b/tests/test_csrf.py
@@ -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() == ""
diff --git a/tests/test_db_index_url.py b/tests/test_db_index_url.py
new file mode 100644
index 0000000..50f73ce
--- /dev/null
+++ b/tests/test_db_index_url.py
@@ -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
diff --git a/tests/test_db_schema.py b/tests/test_db_schema.py
new file mode 100644
index 0000000..5a4f77c
--- /dev/null
+++ b/tests/test_db_schema.py
@@ -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
diff --git a/tests/test_fts_sanitizer.py b/tests/test_fts_sanitizer.py
new file mode 100644
index 0000000..ad061da
--- /dev/null
+++ b/tests/test_fts_sanitizer.py
@@ -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
diff --git a/tests/test_gateway_limits.py b/tests/test_gateway_limits.py
new file mode 100644
index 0000000..6033c3a
--- /dev/null
+++ b/tests/test_gateway_limits.py
@@ -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
diff --git a/tests/test_handlers_pages.py b/tests/test_handlers_pages.py
new file mode 100644
index 0000000..ab4704c
--- /dev/null
+++ b/tests/test_handlers_pages.py
@@ -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"]
diff --git a/tests/test_handlers_search.py b/tests/test_handlers_search.py
new file mode 100644
index 0000000..f7d2f9e
--- /dev/null
+++ b/tests/test_handlers_search.py
@@ -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", {})
diff --git a/tests/test_handlers_subs.py b/tests/test_handlers_subs.py
new file mode 100644
index 0000000..93ee97d
--- /dev/null
+++ b/tests/test_handlers_subs.py
@@ -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 `` 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
diff --git a/tests/test_handlers_tags.py b/tests/test_handlers_tags.py
new file mode 100644
index 0000000..7ec8f05
--- /dev/null
+++ b/tests/test_handlers_tags.py
@@ -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
diff --git a/tests/test_link_extraction.py b/tests/test_link_extraction.py
new file mode 100644
index 0000000..2d8c741
--- /dev/null
+++ b/tests/test_link_extraction.py
@@ -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 = """
+
+ same
+ cross
+ subdomain
+
+ """
+ _, _, 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 = """
+
+ keep
+ skip
+ skip
+ skip
+ skip
+ skip
+
+ """
+ _, _, 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 = """
+
+ keep
+ skip
+ skip
+ skip
+ skip
+
+ """
+ _, _, 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 = """r """
+ _, _, 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 = """r """
+ _, _, 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 = """
+
+ first
+ second
+ third
+
+ """
+ _, _, 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'{long_text} '
+ _, _, 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 = """
+
+ body content
+ """
+ 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 body
+ """
+ _, _, _, meta = _fetch_with_html(monkeypatch, "https://example.com/", html)
+ assert meta == "open graph fallback"
diff --git a/tests/test_pagination.py b/tests/test_pagination.py
new file mode 100644
index 0000000..05077e0
--- /dev/null
+++ b/tests/test_pagination.py
@@ -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
diff --git a/tests/test_regressions.py b/tests/test_regressions.py
new file mode 100644
index 0000000..f8a5df7
--- /dev/null
+++ b/tests/test_regressions.py
@@ -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"
diff --git a/tests/test_sharing_logic.py b/tests/test_sharing_logic.py
new file mode 100644
index 0000000..c9c06d4
--- /dev/null
+++ b/tests/test_sharing_logic.py
@@ -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
diff --git a/tests/test_ssrf.py b/tests/test_ssrf.py
new file mode 100644
index 0000000..807f9bd
--- /dev/null
+++ b/tests/test_ssrf.py
@@ -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")
diff --git a/tests/test_url_cleanup.py b/tests/test_url_cleanup.py
new file mode 100644
index 0000000..1eef72b
--- /dev/null
+++ b/tests/test_url_cleanup.py
@@ -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"
diff --git a/themes/junimo.html b/themes/junimo.html
index 25688b7..f85e315 100644
--- a/themes/junimo.html
+++ b/themes/junimo.html
@@ -3,15 +3,16 @@
