diff --git a/.dockerignore b/.dockerignore
index 9adf733..031c6d8 100644
--- a/.dockerignore
+++ b/.dockerignore
@@ -1,15 +1,5 @@
__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 09dd2d6..c7c6fa3 100644
--- a/.forgejo/workflows/build.yml
+++ b/.forgejo/workflows/build.yml
@@ -6,76 +6,52 @@ on:
jobs:
build:
- runs-on: ubuntu-latest
+ runs-on: docker
steps:
- name: Checkout code
- uses: https://code.forgejo.org/actions/checkout@v4
+ uses: actions/checkout@v4
- name: Set up Python
+ uses: actions/setup-python@v5
+ with:
+ python-version: '3.11'
+
+ - name: Install dependencies
run: |
- 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
+ pip install -r requirements.txt
+ pip install pyinstaller
- name: Build with PyInstaller
run: |
pyinstaller --onefile --console --name TinyWeb app.py
- - name: Prepare artifact
- run: |
- cp dist/TinyWeb TinyWeb-linux-x64
- chmod +x TinyWeb-linux-x64
- ls -la TinyWeb-linux-x64
+ - name: Upload artifact
+ uses: actions/upload-artifact@v4
+ with:
+ name: TinyWeb-linux-x64
+ path: dist/TinyWeb
+ if-no-files-found: error
- - 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
+ release:
+ needs: build
+ runs-on: docker
+ if: startsWith(github.ref, 'refs/tags/v')
- - 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"
+ steps:
+ - name: Download artifact
+ uses: actions/download-artifact@v4
+ with:
+ name: TinyWeb-linux-x64
- - name: Login to Registry
- run: |
- echo "${{ secrets.REGISTRY_TOKEN }}" | docker login registry.derickphan.com -u _ --password-stdin
+ - name: Make executable
+ run: chmod +x TinyWeb-linux-x64
- - 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
+ - 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') }}
diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml
new file mode 100644
index 0000000..d79edb3
--- /dev/null
+++ b/.github/workflows/build.yml
@@ -0,0 +1,75 @@
+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 713df67..9895f65 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -2,12 +2,6 @@ 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 e44c450..371f15d 100644
--- a/README.md
+++ b/README.md
@@ -12,33 +12,9 @@ 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 [Releases](https://git.derickphan.com/lichenblankie/tinyweb/releases) page:
+Download the latest release for your platform from the [GitHub Releases](https://github.com/anomalyco/tinyweb/releases) page:
| Platform | File |
|----------|------|
@@ -48,56 +24,8 @@ Download the latest release for your platform from the [Releases](https://git.de
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 |
@@ -109,37 +37,13 @@ Your data is stored in `~/.tinyweb/`:
This allows your data to persist between upgrades and stay separate from the application.
-### Backups
-
-Back up the whole `~/.tinyweb/` directory periodically. The two files that matter:
-
-- **`tinyweb_identity`** is your permanent mesh identity. If you lose it, your destination hash changes and every subscriber has to re-subscribe to the new one. Keep it somewhere you trust; the file is `0600` by default.
-- **`index.db`** is your full reading history — every page, note, tag, and synced remote page. Losing it loses everything you've curated.
-
-`models/` and `index.hnsw` are re-derivable (the model will re-download, and the HNSW index rebuilds from the database on next startup with semantic search enabled) so they don't need to be backed up.
-
-The `/export` page produces a JSON dump of your pages. It's a migration aid — it doesn't preserve your identity file, your custom template, or subscription state. A full restore needs a copy of `~/.tinyweb/`.
-
-### Docker
-
-Data is stored in the `/data` volume inside the container. Use a volume mount to persist data:
-
-```bash
-docker run -p 8080:8080 -v tinyweb-data:/data registry.derickphan.com/tinyweb:latest
-```
-
-Or with docker-compose (see above) — data persists in the named volume.
-
### Command line options
```bash
-./TinyWeb --version # Show version
-./TinyWeb -p 9000 # Use port 9000 instead of default 8080
-./TinyWeb --bind 0.0.0.0 # Expose the web UI to your LAN (see warning below)
+./TinyWeb --version # Show version
+./TinyWeb -p 9000 # Use port 9000 instead of default 8080
```
-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
@@ -147,7 +51,7 @@ pip install -r requirements.txt
python app.py
```
-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.
+This starts the Reticulum server and an HTTP gateway on `http://localhost:8080`. Open it in your browser.
Your destination hash is printed on startup — share it with friends so they can subscribe to your index.
@@ -182,9 +86,7 @@ themes/ — Saved HTML templates (e.g. kodama.html)
## Security
-**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:
+TinyWeb includes several 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
@@ -194,23 +96,6 @@ Other 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 b1c4fe6..8302d91 100644
--- a/app.py
+++ b/app.py
@@ -8,8 +8,7 @@ from http.server import HTTPServer
from db import init_db, get_setting, set_setting
from handlers import dispatch_request
-import gateway
-from gateway import GatewayState, GatewayHandler
+from gateway import GatewayState, GatewayHandler, GATEWAY_PORT
APP_NAME = "tinyweb"
ASPECTS = ["server"]
@@ -25,13 +24,13 @@ def get_transport_config():
return host, int(port)
-def find_available_port(start=8080, max_attempts=20, host="127.0.0.1"):
+def find_available_port(start=8080, max_attempts=20):
"""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((host, port))
+ s.bind(("0.0.0.0", port))
return port
except OSError:
continue
@@ -72,62 +71,30 @@ 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, bind_host="127.0.0.1"):
+def start_gateway(reticulum):
GatewayState.reticulum = reticulum
GatewayState.local_dispatch = dispatch_request
- server = HTTPServer((bind_host, gateway.GATEWAY_PORT), GatewayHandler)
+ server = HTTPServer(("0.0.0.0", GATEWAY_PORT), GatewayHandler)
thread = threading.Thread(target=server.serve_forever, daemon=True)
thread.start()
-def _config_settings_match(config_file, desired_host, desired_port):
- """Check if existing config transport and LoRa settings match desired values."""
+def _transport_settings_match(config_file, desired_host, desired_port):
+ """Check if existing config transport settings match desired values."""
import configparser
try:
config = configparser.ConfigParser()
config.read(config_file)
- # 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
+ 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)
except Exception:
pass
return False
@@ -143,60 +110,13 @@ 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):
- 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."
- )
+ if _transport_settings_match(config_file, transport_host, transport_port):
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"""{managed_sentinel}
-[reticulum]
+ f.write(f"""[reticulum]
enable_transport = False
share_instance = No
@@ -207,7 +127,13 @@ def ensure_rns_config(config_dir, transport_host=None, transport_port=None):
[[Default Interface]]
type = AutoInterface
enabled = Yes
-{tcp_block}{lora_block}""")
+
+ [[TCP Transport]]
+ type = TCPClientInterface
+ enabled = yes
+ target_host = {transport_host}
+ target_port = {transport_port}
+""")
print(f"Created Reticulum config at {config_file}")
@@ -233,20 +159,15 @@ 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
- gateway.GATEWAY_PORT = find_available_port(port, host=bind_host)
+ import gateway
+ gateway.GATEWAY_PORT = find_available_port(port)
init_db()
transport_host = get_setting("transport_host", DEFAULT_TRANSPORT_HOST)
@@ -275,15 +196,10 @@ def main():
time.sleep(2)
destination.announce()
set_setting("dest_hash", destination.hash.hex())
- start_gateway(reticulum, bind_host=bind_host)
+ start_gateway(reticulum)
print(f"TinyWeb running!")
- 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"Open http://localhost:{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
deleted file mode 100644
index 9a2f26e..0000000
--- a/conftest.py
+++ /dev/null
@@ -1,128 +0,0 @@
-"""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 8a645ab..295da86 100644
--- a/db.py
+++ b/db.py
@@ -70,16 +70,10 @@ def clean_url(url):
# Prefer https
scheme = "https" if parsed.scheme in ("http", "https") else parsed.scheme
- # Normalize hostname: lowercase, strip www (only if non-www resolves)
+ # Normalize hostname: lowercase, strip www.
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
@@ -103,7 +97,7 @@ def clean_url(url):
_pool = []
_pool_lock = __import__("threading").Lock()
-_POOL_SIZE = 16
+_POOL_SIZE = 4
def get_db():
@@ -123,14 +117,6 @@ 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)
@@ -285,15 +271,8 @@ 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()
@@ -307,16 +286,6 @@ 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:
@@ -420,7 +389,7 @@ def index_url(url, note="", reticulum_dest=""):
(page_id, href, label),
)
db.commit()
- if get_setting("semantic_search", "0") == "1":
+ if get_setting("semantic_search", "1") == "1":
try:
from embeddings import store_embeddings
store_embeddings(page_id, title, body, db)
diff --git a/embeddings.py b/embeddings.py
index 03f6f13..302a31f 100644
--- a/embeddings.py
+++ b/embeddings.py
@@ -233,49 +233,24 @@ def embed(texts, is_query=False):
"token_type_ids": token_type_ids,
},
)
+ # CLS token pooling — take the first token's hidden state
emb = outputs[0][:, 0, :]
all_embeddings.append(emb)
embeddings = np.concatenate(all_embeddings, axis=0)
+ # L2 normalize
norms = np.linalg.norm(embeddings, axis=1, keepdims=True)
norms = np.maximum(norms, 1e-12)
embeddings = embeddings / norms
- return _maybe_compress(embeddings.astype(np.float32))
-
-
-def _maybe_compress(embeddings):
- """Compress embeddings to float16 if compression is enabled."""
- try:
- from db import get_setting
- if get_setting("compress_embeddings", "0") == "1":
- return embeddings.astype(np.float16)
- except Exception:
- pass
- return embeddings
-
-
-def _decompress(embeddings):
- """Decompress float16 embeddings to float32 if needed."""
- if embeddings.dtype == np.float16:
- return embeddings.astype(np.float32)
- return embeddings
-
-
-def _blob_to_vec(buf):
- """Decode a stored embedding blob to a float32 vector, inferring dtype from length."""
- if len(buf) == DIMS * 2:
- return np.frombuffer(buf, dtype=np.float16).astype(np.float32)
- return np.frombuffer(buf, dtype=np.float32)
+ return embeddings.astype(np.float32)
# ---------------------------------------------------------------------------
# HNSW index management
# ---------------------------------------------------------------------------
-BATCH_SIZE = 50000
-
def build_index(db=None):
- """Load all embeddings from chunks table and build HNSW index in batches."""
+ """Load all embeddings from chunks table and build HNSW index."""
import hnswlib
global _hnsw_index, _hnsw_ids
@@ -283,47 +258,29 @@ def build_index(db=None):
own_db = db is None
if own_db:
db = get_db()
-
try:
- 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)
+ rows = db.execute("SELECT id, embedding FROM chunks ORDER BY id").fetchall()
finally:
if own_db:
return_db(db)
- if not all_ids:
- with _hnsw_lock:
+ with _hnsw_lock:
+ if not rows:
_hnsw_index = None
_hnsw_ids = []
- return
+ return
- matrix = np.stack(all_embeddings)
- n = len(all_ids)
- ids = all_ids
+ 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)
- 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)
+ 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
- with _hnsw_lock:
_hnsw_index = index
_hnsw_ids = ids
@@ -362,8 +319,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 = []
@@ -386,7 +343,6 @@ 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 1f49fcb..e4a9719 100755
--- a/entrypoint.sh
+++ b/entrypoint.sh
@@ -30,5 +30,4 @@ EOF
fi
export RNS_CONFIG_DIR="$CONFIG_DIR"
-# 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 "$@"
+exec python app.py
diff --git a/gateway.py b/gateway.py
index d07924d..a13816f 100644
--- a/gateway.py
+++ b/gateway.py
@@ -1,4 +1,3 @@
-import re
import sys
import time
import threading
@@ -10,7 +9,6 @@ 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:
@@ -73,18 +71,8 @@ class GatewayHandler(BaseHTTPRequestHandler):
body = {}
if method == "POST":
- 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")
+ length = int(self.headers.get("Content-Length", 0))
+ raw = self.rfile.read(length).decode()
body = parse_qs(raw)
# Parse cookies
@@ -135,14 +123,6 @@ 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()
@@ -164,14 +144,7 @@ class GatewayHandler(BaseHTTPRequestHandler):
self._forward("POST")
def log_message(self, format, *args):
- 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}")
+ print(f"[Gateway] {args[0]}")
def main():
diff --git a/handlers.py b/handlers.py
index e47520b..ae7f484 100644
--- a/handlers.py
+++ b/handlers.py
@@ -50,18 +50,14 @@ 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 (operators, column filter colon) to prevent injection
- cleaned = re.sub(r'["\'\(\)\*\+\-\^~:]', '', w).strip()
+ # Strip FTS5 special characters to prevent injection
+ cleaned = re.sub(r'["\'\(\)\*\+\-\^~]', '', w).strip()
if not cleaned:
continue
if cleaned.lower() in _STOPWORDS:
continue
- # 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:
+ if i == len(words) - 1:
# Prefix match on the last token for partial word matching
tokens.append(f"{cleaned}*")
else:
@@ -178,11 +174,6 @@ 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 ---
@@ -214,7 +205,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", "0") == "1":
+ if get_setting("semantic_search", "1") == "1":
try:
from embeddings import hybrid_search
use_reranker = get_setting("use_reranker", "1") == "1"
@@ -254,7 +245,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}'
@@ -285,7 +276,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 = (
@@ -320,8 +311,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(clean_url(r["url"]))}) '
+ f'
{esc(r["title"])} '
+ f'{note_html} ({esc(r["url"])}) '
)
remote_html += (
f'
'
@@ -334,18 +325,6 @@ 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}'
@@ -381,8 +359,7 @@ def handle_add_form(msg="", action_type="index"):
f'{_csrf_field()}'
f'tag: private to exclude from sharing index '
f""
f"{msg}
"
@@ -444,7 +421,7 @@ def handle_add_submit(body):
f'Title: Description: save manually '
f""
f'back '
@@ -462,8 +439,8 @@ def handle_add_manual_submit(body):
if not url:
return handle_add_form("URL is required.")
- if not manual_title:
- return handle_add_form("Title is required for manual entry.")
+ if not manual_title or not manual_desc:
+ return handle_add_form("Title and description are required for manual entry.")
db = get_db()
try:
@@ -486,7 +463,7 @@ def handle_add_manual_submit(body):
db.commit()
# Generate embeddings for this page (if semantic search is enabled)
- if get_setting("semantic_search", "0") == "1":
+ if get_setting("semantic_search", "1") == "1":
try:
from embeddings import store_embeddings
# Pass the page_id, title, description, and db connection
@@ -496,7 +473,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)
@@ -522,9 +499,8 @@ def handle_pages(query=None):
tag_links = " ".join(f'[{esc(t)}] ' for t in tags)
tags_html = f' {tag_links}'
items += (
- f'({esc(r["url"])} ) '
+ f'{esc(r["title"])}{note_html}{tags_html} '
+ f'({esc(r["url"])} ) '
f'edit '
f'remove '
)
@@ -533,112 +509,13 @@ def handle_pages(query=None):
return _respond(
f"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:
@@ -662,8 +539,7 @@ 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}
"
@@ -685,7 +561,6 @@ def handle_edit_submit(page_id, body):
)
_set_page_tags(page_id, tags, db)
- _cleanup_orphaned_tags(db)
db.commit()
@@ -721,7 +596,6 @@ 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)
@@ -744,19 +618,10 @@ def handle_bookmark(query):
return _text_response(msg, headers={"Access-Control-Allow-Origin": "*"})
-MAX_EXPORT = 10000
-
-def handle_export(query=None):
- try:
- batch = int((query or {}).get("batch", ["0"])[0])
- except (TypeError, ValueError):
- batch = 0
+def handle_export():
db = get_db()
try:
- rows = db.execute(
- "SELECT url, title, note FROM pages ORDER BY id LIMIT ? OFFSET ?",
- (MAX_EXPORT, batch * MAX_EXPORT),
- ).fetchall()
+ rows = db.execute("SELECT url, title, note FROM pages ORDER BY id").fetchall()
finally:
return_db(db)
data = [{"url": r["url"], "title": r["title"], "note": r["note"]} for r in rows]
@@ -813,33 +678,14 @@ 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 "
@@ -848,57 +694,13 @@ 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" share your site list publicly at /api/sitesmesh network "
- f"Choose how to connect to the mesh. You can enable both for maximum reach.
"
- f"internet "
- f'Reach peers anywhere online. '
+ f"
Connect to a Reticulum transport node to reach other peers.
"
f"
Default: reticulum.derickphan.com:4242 "
- f'
'
- f'
'
- f'
discover more nodes
'
- f'
LoRa "
- f'Reach nearby peers off-grid with an RNode . '
- f'Serial port: advanced radio settings '
- f'
discover more nodes
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 "
@@ -925,16 +724,10 @@ 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,
@@ -945,14 +738,8 @@ 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 "")
@@ -960,20 +747,10 @@ 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.")
@@ -1015,22 +792,6 @@ 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, '
@@ -1090,7 +851,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)
@@ -1103,125 +864,6 @@ 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(
@@ -1232,8 +874,23 @@ def handle_api_sites(query=None):
since = (query or {}).get("since", [""])[0].strip()
db = get_db()
try:
- sites = _shared_sites(db, since=since)
- all_urls = _shared_all_urls(db) if not since else None
+ 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
finally:
return_db(db)
data = {"name": get_site_name(), "sites": sites}
@@ -1242,9 +899,6 @@ 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:
@@ -1253,54 +907,30 @@ 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'{sync_btn}'
- f'
'
+ f'
'
- f'
'
f'
'
f'
syncing... ' if any_syncing else 'sync all '
listing = (
f'{cards}'
f''
+ f'{_csrf_field()}sync all '
)
return _respond(
f"subscriptions "
@@ -1344,20 +974,18 @@ 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 LIMIT ?", (MAX_BROWSE,)).fetchall())
+ local_urls = set(r["url"] for r in db.execute("SELECT url FROM pages").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 = ? LIMIT ?",
- (sub_id, MAX_BROWSE),
+ "SELECT url, title, note, tags FROM remote_pages WHERE subscription_id = ?",
+ (sub_id,),
).fetchall()
finally:
return_db(db)
@@ -1427,7 +1055,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 LIMIT ?", (MAX_BROWSE,)).fetchall())
+ local_urls = set(r["url"] for r in db.execute("SELECT url FROM pages").fetchall())
urls = [r["url"] for r in remote_rows if r["url"] not in local_urls]
else:
urls = body.get("urls", [])
@@ -1459,15 +1087,13 @@ def handle_subscription_pick(body):
return handle_subscriptions(f"Imported {imported} page(s). {errors} error(s).")
-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")
+def handle_subscription_sync(sub_id):
db = get_db()
try:
sub = db.execute("SELECT * FROM subscriptions WHERE id = ?", (sub_id,)).fetchone()
if not sub:
- set_setting(f"sync_status_{sub_id}", "error:Subscription not found.")
- return
+ return handle_subscriptions("Subscription not found.")
+ # Use last_sync for delta sync if available
since = sub["last_sync"].replace(" ", "T") if sub["last_sync"] else ""
try:
data = fetch_remote_sites(sub["dest_hash"], since=since)
@@ -1475,12 +1101,11 @@ def _sync_subscription(sub_id):
all_urls = data.get("all_urls")
remote_name = data.get("name", sub["name"])
except PermissionError:
- 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
+ return handle_subscriptions("That instance has sharing disabled.")
+ except Exception:
+ return handle_subscriptions("Could not sync with that instance.")
+ # 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,)
@@ -1490,6 +1115,7 @@ def _sync_subscription(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:
@@ -1499,7 +1125,8 @@ def _sync_subscription(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),
)
- if get_setting("semantic_search", "0") == "1":
+ # Embed remote page for semantic search
+ if get_setting("semantic_search", "1") == "1":
try:
from embeddings import store_remote_embeddings
rp_id = db.execute(
@@ -1515,22 +1142,9 @@ def _sync_subscription(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)
-
-
-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")
+ return handle_subscriptions(f"Synced {synced} site(s) from {esc(remote_name)}.")
def handle_subscription_autosync(sub_id):
@@ -1562,15 +1176,53 @@ 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:
- 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")
+ 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).")
# --- Reindex (semantic search) ---
@@ -1580,7 +1232,7 @@ _reindex_thread = None
def handle_reindex_form():
- if get_setting("semantic_search", "0") != "1":
+ if get_setting("semantic_search", "1") != "1":
return _respond(
f"semantic search index "
f"Semantic search is disabled. Enable it in settings to use embeddings.
"
@@ -1667,12 +1319,10 @@ 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(query)
+ return handle_export()
elif path == "/import":
return handle_import_form()
elif path == "/tags":
@@ -1694,8 +1344,6 @@ 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/"):
@@ -1709,10 +1357,6 @@ 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":
@@ -1753,7 +1397,8 @@ def dispatch_request(data):
resp["headers"]["Content-Security-Policy"] = (
"default-src 'self'; "
"script-src 'self' 'unsafe-inline'; "
- "style-src 'self' 'unsafe-inline'; "
+ "style-src 'self' 'unsafe-inline' https://fonts.googleapis.com; "
+ "font-src 'self' https://fonts.gstatic.com; "
"img-src * data:; "
"frame-ancestors 'none'; "
"form-action 'self'; "
diff --git a/pytest.ini b/pytest.ini
deleted file mode 100644
index 16d6cc5..0000000
--- a/pytest.ini
+++ /dev/null
@@ -1,5 +0,0 @@
-[pytest]
-testpaths = tests
-python_files = test_*.py
-filterwarnings =
- ignore::DeprecationWarning
diff --git a/requirements-dev.txt b/requirements-dev.txt
deleted file mode 100644
index 26b77f6..0000000
--- a/requirements-dev.txt
+++ /dev/null
@@ -1,2 +0,0 @@
--r requirements.txt
-pytest
diff --git a/rns_client.py b/rns_client.py
index 98df406..dbc0af5 100644
--- a/rns_client.py
+++ b/rns_client.py
@@ -1,15 +1,9 @@
-import json
import time
import RNS
APP_NAME = "tinyweb"
ASPECTS = ["server"]
-
-# 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},
-]
+REQUEST_TIMEOUT = 30
def fetch_remote_sites(dest_hash_hex, since=""):
@@ -17,40 +11,18 @@ 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 < timeouts["path"]:
- time.sleep(poll)
- elapsed += poll
+ while not RNS.Transport.has_path(dest_hash) and elapsed < 15:
+ time.sleep(0.5)
+ elapsed += 0.5
if not RNS.Transport.has_path(dest_hash):
- raise ConnectionError(
- f"Could not find path to {dest_hash_hex} ({timeouts['path']}s timeout)"
- )
+ raise ConnectionError(f"Could not find path to {dest_hash_hex}")
server_identity = RNS.Identity.recall(dest_hash)
if server_identity is None:
@@ -67,16 +39,15 @@ def _fetch(dest_hash_hex, since, timeouts):
# Establish link
link = RNS.Link(destination)
elapsed = 0
- while link.status == RNS.Link.PENDING and elapsed < timeouts["link"]:
- time.sleep(poll)
- elapsed += poll
+ while link.status == RNS.Link.PENDING and elapsed < 15:
+ time.sleep(0.25)
+ elapsed += 0.25
if link.status != RNS.Link.ACTIVE:
- raise ConnectionError(
- f"Could not establish link to {dest_hash_hex} ({timeouts['link']}s timeout)"
- )
+ raise ConnectionError(f"Could not establish link to {dest_hash_hex}")
try:
+ # Request /api/sites
query = {"since": [since]} if since else {}
request_data = {
"method": "GET",
@@ -86,14 +57,13 @@ def _fetch(dest_hash_hex, since, timeouts):
"gateway_host": "",
}
- req_timeout = timeouts["request"]
- receipt = link.request("/tinyweb", data=request_data, timeout=req_timeout)
+ receipt = link.request("/tinyweb", data=request_data, timeout=REQUEST_TIMEOUT)
elapsed = 0
done = (RNS.RequestReceipt.READY, RNS.RequestReceipt.DELIVERED, RNS.RequestReceipt.FAILED)
- while receipt.get_status() not in done and elapsed < req_timeout:
- time.sleep(poll)
- elapsed += poll
+ while receipt.get_status() not in done and elapsed < REQUEST_TIMEOUT:
+ time.sleep(0.5)
+ elapsed += 0.5
if receipt.get_status() in (RNS.RequestReceipt.READY, RNS.RequestReceipt.DELIVERED):
resp = receipt.get_response()
@@ -101,10 +71,9 @@ def _fetch(dest_hash_hex, since, timeouts):
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 ({req_timeout}s timeout)"
- )
+ raise ConnectionError(f"Request failed or timed out")
finally:
link.teardown()
diff --git a/templates.py b/templates.py
index 0dd9975..48beace 100644
--- a/templates.py
+++ b/templates.py
@@ -7,13 +7,13 @@ def esc(s):
-DEFAULT_TEMPLATE = "\n\n{name} search | browse '
' | tags | subscriptions '
diff --git a/tests/test_csrf.py b/tests/test_csrf.py
deleted file mode 100644
index 43b4487..0000000
--- a/tests/test_csrf.py
+++ /dev/null
@@ -1,60 +0,0 @@
-"""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
deleted file mode 100644
index 50f73ce..0000000
--- a/tests/test_db_index_url.py
+++ /dev/null
@@ -1,155 +0,0 @@
-"""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
deleted file mode 100644
index 5a4f77c..0000000
--- a/tests/test_db_schema.py
+++ /dev/null
@@ -1,90 +0,0 @@
-"""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
deleted file mode 100644
index ad061da..0000000
--- a/tests/test_fts_sanitizer.py
+++ /dev/null
@@ -1,113 +0,0 @@
-"""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
deleted file mode 100644
index 6033c3a..0000000
--- a/tests/test_gateway_limits.py
+++ /dev/null
@@ -1,164 +0,0 @@
-"""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
deleted file mode 100644
index ab4704c..0000000
--- a/tests/test_handlers_pages.py
+++ /dev/null
@@ -1,174 +0,0 @@
-"""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
deleted file mode 100644
index f7d2f9e..0000000
--- a/tests/test_handlers_search.py
+++ /dev/null
@@ -1,63 +0,0 @@
-"""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
deleted file mode 100644
index 93ee97d..0000000
--- a/tests/test_handlers_subs.py
+++ /dev/null
@@ -1,112 +0,0 @@
-"""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
deleted file mode 100644
index 7ec8f05..0000000
--- a/tests/test_handlers_tags.py
+++ /dev/null
@@ -1,101 +0,0 @@
-"""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
deleted file mode 100644
index 2d8c741..0000000
--- a/tests/test_link_extraction.py
+++ /dev/null
@@ -1,138 +0,0 @@
-"""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
deleted file mode 100644
index 05077e0..0000000
--- a/tests/test_pagination.py
+++ /dev/null
@@ -1,58 +0,0 @@
-"""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
deleted file mode 100644
index f8a5df7..0000000
--- a/tests/test_regressions.py
+++ /dev/null
@@ -1,107 +0,0 @@
-"""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
deleted file mode 100644
index c9c06d4..0000000
--- a/tests/test_sharing_logic.py
+++ /dev/null
@@ -1,38 +0,0 @@
-"""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
deleted file mode 100644
index 807f9bd..0000000
--- a/tests/test_ssrf.py
+++ /dev/null
@@ -1,64 +0,0 @@
-"""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
deleted file mode 100644
index 1eef72b..0000000
--- a/tests/test_url_cleanup.py
+++ /dev/null
@@ -1,101 +0,0 @@
-"""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 f85e315..25688b7 100644
--- a/themes/junimo.html
+++ b/themes/junimo.html
@@ -3,16 +3,15 @@
