opencode/NSCT---Neutral-Search-Crawler-Tool
Public Access
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Stage 0: Repository und Architekturgrundlage

Browse Source 
- Pyproject.toml mit FastAPI, Pydantic v2, SQLAlchemy, httpx, asyncio,
  BeautifulSoup4, selectolax, trafilatura, uvicorn, pytest-asyncio
- Multi-stage Dockerfile (Python 3.12-slim, Non-Root-User nsct)
- docker-compose.yml (nsct-api + postgres + optional searxng)
- .env.example mit allen Config-Parametern
- Config-System: AppSettings mit LLMConfig, VisionConfig, AudioConfig,
  DatabaseConfig — komplett aus Environment, keine Hardcodes
- Strukturiertes Logging mit research_id/llm_request_id Tracking
- Pydantic v2 Schemas: SearchQuery, Source, Claim, EvidenceRelation,
  CitationEdge, ResearchReport
- SQLAlchemy 2.0 Declarative Models + async Engine Factory
- SSRF-Schutz: URL-Validation, IP-Blocklist (RFC1918, Cloud Metadata,
  file://, ftp://)
- Provider-Interfaces: LLMProvider, VisionProvider, AudioProvider,
  SearchProvider, ContentFetcher als ABCs
- Health-Endpoints: /health, /ready (LLM-Connect-Test), /providers
- FastAPI App mit CORS, lifespan (LLM Pre-Flight)
- CLI-Stub mit Entry-Points: nsct, nsct-core, nsct-api
- 6 Test-Cases: /health, /ready, /providers + No-Secrets-Test
- Vollständige Dokumentation: README, ARCHITECTURE, SECURITY,
  METHODOLOGY, API, DEPLOYMENT
- .gitignore (Python, Docker, IDE, .env)
This commit is contained in:
NSCT Agent
2026-08-23 11:33:45 +00:00
commit e9410be941
28 changed files with 4192 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
.env.example Normal file
View File

@@ -0,0 +1,34 @@
# ============================================================
# NSCT — Environment Variables (copy to .env and fill in)
# ============================================================
# ---------- LLM ----------
# OpenAI-compatible endpoint for the primary LLM worker
NSCT_LLM_BASE_URL=http://localhost:8030/openai/v1
NSCT_LLM_MODEL=Qwen3.6-6-35B
NSCT_LLM_MAX_CONCURRENCY=3
# Optional API key for the LLM endpoint (provider-specific header)
NSCT_LLM_API_KEY=
# ---------- Vision (image analysis) ----------
NSCT_VISION_BASE_URL=http://localhost:8030/openai/visual/v1
NSCT_VISION_MODEL=Qwen2-VL-3B
NSCT_VISION_API_KEY=
# ---------- Audio (speech / transcription) ----------
NSCT_AUDIO_BASE_URL=http://localhost:8030/hermes-audio
NSCT_AUDIO_MODEL=default
NSCT_AUDIO_API_KEY=
# ---------- PostgreSQL ----------
NSCT_DB_URL=postgresql+asyncpg://nsct:nsct_secret@localhost:5432/nsct
POSTGRES_USER=nsct
POSTGRES_PASSWORD=nsct_secret
POSTGRES_DB=nsct
# ---------- SearXNG (optional search backend) ----------
NSCT_SEARXNG_BASE_URL=http://localhost:8888/
# ---------- General ----------
NSCT_DEBUG=false

52
.gitignore vendored Normal file
View File

@@ -0,0 +1,52 @@
# Python
__pycache__/
*.py[cod]
*.so
*.egg-info/
dist/
build/
*.egg
.eggs/
# Virtual environments
venv/
.venv/
env/
.env/
# IDE
.idea/
.vscode/
*.swp
*.swo
*~
# Environment files (never commit secrets)
.env
.env.*
!.env.example
# Docker
*.pid
# Coverage
htmlcov/
.coverage
.coverage.*
coverage.xml
# OS
.DS_Store
Thumbs.db
# Logs
*.log
logs/
# Local data
nsct_data/
*.sqlite3
*.sqlite
# Node (if any)
node_modules/

157
API.md Normal file
View File

@@ -0,0 +1,157 @@
# NSCT — API-Referenz
> **Hinweis:** Diese API-Referenz beschreibt nur die Endpunkte
> von Stage 0. Die eigentliche Evidence-Pipeline (Suche,
> Analyse, Report) wird in späteren Stages implementiert.
## Server-Adresse
```
http://localhost:8080
```
Nach `docker compose up`.
---
## System-Endpunkte
### GET /health
**Beschreibung:** Basic health check — gibt den Status und die Version zurück.
Wird von Docker Healthchecks und Load Balancern verwendet.
**Request:**
```
GET /health
```
**Response (200 OK):**
```json
{
"status": "ok",
"version": "0.1.0"
}
```
---
### GET /ready
**Beschreibung:** Readiness check — prüft die Erreichbarkeit von
Downstream-Services (LLM-Provider, ggf. Datenbank).
**Request:**
```
GET /ready
```
**Response (200 OK):**
Wenn bereit:
```json
{
"status": "ready",
"llm": "ok",
"database": "unknown",
"llm_model": "Qwen3.6-6-35B"
}
```
Wenn nicht bereit:
```json
{
"status": "not_ready",
"llm": "error:Connection refused",
"database": "unknown"
}
```
**Mögliche `status`-Werte:**
- `ready` — Alle geprüften Services sind erreichbar
- `not_ready` — Mindestens ein Service ist nicht erreichbar
**Mögliche `llm`-Werte:**
- `ok` — LLM-Provider ist erreichbar
- `error:<detail>` — Fehlerbeschreibung (HTTP-Statuscode oder Exception-Typ)
---
### GET /providers
**Beschreibung:** Listet die konfigurierten Provider ohne Secrets.
**Request:**
```
GET /providers
```
**Response (200 OK):**
```json
{
"llm": {
"available": true,
"model": "Qwen3.6-6-35B",
"max_concurrency": 3
},
"vision": {
"available": true,
"model": "Qwen2-VL-3B"
},
"audio": {
"available": true,
"model": "default"
}
}
```
**Falls kein Provider konfiguriert:**
```json
{
"llm": { "available": false },
"vision": { "available": false },
"audio": { "available": false }
}
```
**Wichtig:** Dieser Endpoint gibt **keinerlei** Secrets, API-Keys
oder senssible Konfigurationswerte zurück.
---
## Entwicklung (nur mit NSCT_DEBUG=true)
### GET /docs
Swagger UI mit interaktiver API-Documentation.
```bash
curl http://localhost:8080/docs
```
### GET /redoc
ReDoc-Generierung.
```bash
curl http://localhost:8080/redoc
```
---
## Error Response Format
Alle Fehler folgen einem konsistenten Format:
```json
{
"detail": "Fehlerbeschreibung"
}
```
---
## Authentifizierung
Stage 0: Keine Authentifizierung. Dies wird in späteren Stages
hinzugefügt (API-Key, OAuth, etc.).

168
ARCHITECTURE.md Normal file
View File

@@ -0,0 +1,168 @@
# NSCT — Architekturdokumentation
## 1. Überblick
NSCT (Neutral Search Crawler Tool) ist ein modulares, containerisiertes
Recherchesystem. Es durchsucht Webquellen, extrahiert Inhalte, analysiert
Behauptungen (Claims) und erzeugt einen neutralen, quellengestützten Bericht.
**Kernprinzip:** Webcontent ist DATA, keine INSTRUCTION. Der LLM verarbeitet
nur strukturierte Daten, niemals rohen Webcontent direkt als Prompt.
## 2. Architektur-Diagramm
```
┌─────────────────────────────────────────────────────────────────────┐
│ Control Plane │
│ ┌──────────┐ ┌───────────┐ ┌────────────┐ ┌─────────────────┐ │
│ │ CLI │ │ REST API │ │ WebSocket │ │ Event Bus │ │
│ └────┬─────┘ └─────┬─────┘ └─────┬──────┘ └────────┬────────┘ │
│ │ │ │ │ │
│ └──────────────┴─────────────┴───────────────────┘ │
│ │
│ ┌───────────────────────Evidence Pipeline───────────────────────┐ │
│ │ │ │
│ │ ┌─────────┐ ┌──────────┐ ┌───────────┐ ┌───────────┐ │ │
│ │ │ Search │──▶│ Fetch │──▶│ Extract │──▶│ Classify │ │ │
│ │ │ (SearX │ │ (HTTP/ │ │(Trafilat │ │(LLM/Rule)│ │ │
│ │ │ NG) │ │ Playwright)│ │ura/BS4) │ │ │ │ │
│ │ └─────────┘ └──────────┘ └───────────┘ └─────┬─────┘ │ │
│ │ │ │ │
│ │ ┌─────────┐ ┌───────────┐ ┌────────────┐ ┌───┴─────┐ │ │
│ │ │ Index │◀──│ Compare │◀──│ Normalise │◀──│ Claim │ │ │
│ │ │ & Store │ │ (LLM) │ │ & Parse │ │ Ext. │ │ │
│ │ └─────────┘ └─────┬─────┘ └────────────┘ └─────────┘ │ │
│ │ │ │ │
│ │ ▼ │ │
│ │ ┌──────────────┐ │ │
│ │ │ Report Gen. │ │ │
│ │ │ (LLM) │ │ │
│ │ └──────┬───────┘ │ │
│ │ ▼ │ │
│ │ ┌──────────────┐ │ │
│ │ │ Evidence DB │ │ │
│ │ └──────────────┘ │ │
│ └──────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────┘
```
## 3. Datenfluss
```
Query ──▶ SearchProvider.search() ──▶ List[SearchResult]
│
│ (for each result)
│
▼
ContentFetcher.fetch(url) ──▶ Source document (parsed)
│
│ (LLM: extract claims from source)
│
▼
Claim[] ──▶ Normalization ──▶ NormalizedClaim[]
│
│ (LLM: compare claims pairwise)
│
▼
EvidenceRelation[] ──▶ Agreement matrix
│
│ (LLM: synthesize report)
│
▼
ResearchReport (Summary + Findings + Disagreements + Uncertainties)
```
## 4. Module
### 4.1 `src/nsct/api/` — REST API
- **main.py**: FastAPI application factory, CORS, lifespan hooks
- **health.py**: `/health`, `/ready`, `/providers` endpunkte
- **Pattern:** Each router is a separate file; mounted in `main.py`
### 4.2 `src/nsct/models/` — Pydantic v2 Schemas
- **schemas.py**: Data transfer objects (SearchQuery, Source, Claim, etc.)
- Pure Pydantic — no database or API coupling
### 4.3 `src/nsct/storage/` — Database Layer
- **models.py**: SQLAlchemy 2.0 declarative models
- **engine.py**: Async engine factory with pool management
- Uses `asyncpg` for PostgreSQL
### 4.4 `src/nsct/providers/` — Provider Interfaces
- Abstract base classes (LLMProvider, VisionProvider, SearchProvider,
ContentFetcher, AudioProvider)
- **No concrete implementations in Stage 0** — just protocols
- All providers accept structured data, never raw web content
### 4.5 `src/nsct/security/` — Security
- **policy.py**: SSRF protection, URL validation, IP blocklisting
- Called before every outbound HTTP request
### 4.6 `src/nsct/config.py` — Configuration
- Pydantic BaseSettings — all values from environment
- Zero hard-coded secrets or URLs
- `AppSettings.from_env()` creates the root configuration
### 4.7 `src/nsct/logging_config.py` — Structured Logging
- JSON-formatted log output
- Per-request context tracking (`research_id`, `llm_request_id`)
- Global context dict merged into every log record
## 5. Interfaces / Protocols
```python
# src/nsct/providers/__init__.py
class LLMProvider(ABC):
async def generate(prompt, system_prompt=None, ...) -> LLMResponse: ...
class VisionProvider(ABC):
async def analyze_image(image_bytes, prompt) -> VisionResponse: ...
class AudioProvider(ABC):
async def transcribe(audio_bytes) -> AudioResponse: ...
class SearchProvider(ABC):
async def search(query, language, limit) -> list[SearchResult]: ...
class ContentFetcher(ABC):
async def fetch(url, **kwargs) -> dict: ...
```
Alle konkreten Implementationen (SearXNG, OpenAI, Trafilatura, etc.)
müssen diese Interfaces implementieren — das ermöglicht den Wechsel
von Providern ohne Codeänderung im Core.
## 6. Sicherheitsarchitektur
- **Control Plane vs. Evidence Plane:** Der LLM verarbeitet nur strukturierte
Daten (Claims, NormalizedClaims, EvidenceRelations), niemals rohen Webcontent.
- **SSRF-Schutz:** Jede outbound-URL wird vor dem Request durch `validate_url()` geprüft.
- **Prompt-Isolation:** Webcontent wird niemals direkt als Prompt text eingebettet.
Stattdessen wird er in strukturierte JSON-Objekte serialisiert.
- **Non-Root-Docker:** Der Container läuft als nicht-root User `nsct`.
- **Read-Only-Filesystem:** Wo möglich (`read_only: true` + `tmpfs`).
## 7. Datenmodell
### Core Tables
| Table | Beschreibung |
|-------|-------------|
| `search_queries` | Research-Aufträge mit Query, Purpose, Language |
| `sources` | Extrahierte Webquellen mit Metadaten |
| `claims` | Behauptungen aus Quellen mit Typ, Konfidenz, Evidence |
| `evidence_relations` | Vergleichsergebnisse zwischen Claims |
| `citation_edges` | Quelle-zu-Quelle Referenzen |
| `research_reports` | Aggregierte Forschungsberichte |
## 8. Nicht-funktionale Anforderungen
| Kriterium | Anforderung |
|-----------|------------|
| Concurrency | Async I/O konsequent (async/await everywhere) |
| Database | PostgreSQL 16+, asyncpg, SQLAlchemy 2.0 |
| Logging | Strukturiert (JSON), jede Anfrage tracebar |
| Config | Environment-only, keine Config-Dateien |
| Testing | pytest-asyncio, FastAPI TestClient |
| Deployment | Docker Compose, reproduzierbar |

183
DEPLOYMENT.md Normal file
View File

@@ -0,0 +1,183 @@
# NSCT — Deployment-Anleitung
## 1. Voraussetzungen
- **Docker** (≥ 24.0) und **Docker Compose** (≥ 2.23)
- Mindestens 2 GB freier RAM
- Zugang zu einem OpenAI-kompatiblen LLM-Endpunkt
## 2. Konfiguration
### 2.1 Environment-Datei
```bash
cp .env.example .env
```
Trage folgende Werte ein:
```env
# LLM (verpflichtend)
NSCT_LLM_BASE_URL=http://192.168.80.199:8030/openai/v1
NSCT_LLM_MODEL=Qwen3.6-6-35B
NSCT_LLM_MAX_CONCURRENCY=3
NSCT_LLM_API_KEY=<dein-api-key>
# Vision (optional)
NSCT_VISION_BASE_URL=http://192.168.80.199:8030/openai/visual/v1
NSCT_VISION_MODEL=Qwen2-VL-3B
# Audio (optional)
NSCT_AUDIO_BASE_URL=http://192.168.80.199:8030/hermes-audio
NSCT_AUDIO_MODEL=default
# PostgreSQL
POSTGRES_USER=nsct
POSTGRES_PASSWORD=<starkes-passwort>
POSTGRES_DB=nsct
NSCT_DB_URL=postgresql+asyncpg://nsct:<passwort>@postgres:5432/nsct
# SearXNG (optional)
NSCT_SEARXNG_BASE_URL=http://searxng:8080
# Debug
NSCT_DEBUG=false
```
## 3. PostgreSQL Setup
### Option A: Docker Compose (empfohlen)
```bash
docker compose up postgres
```
Die Datenbank wird automatisch erstellt. Die Credentials stehen in `.env`.
### Option B: Externe PostgreSQL
1. Erstelle die Datenbank und den User manuell:
```sql
CREATE DATABASE nsct;
CREATE USER nsct WITH ENCRYPTED PASSWORD '<passwort>';
GRANT ALL PRIVILEGES ON DATABASE nsct TO nsct;
```
2. Setze `NSCT_DB_URL` in `.env` auf den externen Connection String.
## 4. Deployment-Schritte
### 4.1 Build und Start
```bash
# Vollständiger Stack (API + PostgreSQL + SearXNG)
docker compose up --build
# Nur API und PostgreSQL (kein SearXNG)
docker compose up --build nsct-api postgres
```
### 4.2 Gesundheitsprüfung
```bash
# Health check
curl http://localhost:8080/health
# Readiness check
curl http://localhost:8080/ready
# Provider-Status
curl http://localhost:8080/providers
```
### 4.3 Logs
```bash
docker compose logs -f nsct-api
```
## 5. Production-Hinweise
### 5.1 Sicherheit
- `.env` **niemals** committen — `.gitignore` behandelt das
- Verwende ein Secrets-Management Tool (Hashicorp Vault, AWS Secrets Manager)
- API-Authentifizierung wird in Stage 2+ implementiert
- Network-Policies für Docker (nur interner Traffic zwischen Services)
### 5.2 Skalierung
- Derzeit: Single-Instance (kein horizontal scaling)
- PostgreSQL: Connection Pooling über `pool_size=10, max_overflow=20`
- LLM: Max. `NSCT_LLM_MAX_CONCURRENCY` parallele Requests (standard: 3)
### 5.3 Datenpersistenz
```yaml
# docker-compose.yml
volumes:
postgres_data: # PostgreSQL Daten
driver: local
nsct_data: # NSCT Runtime-Daten
driver: local
```
Für Production: Verwende ein volumen-Plugin mit Backup-Unterstützung
(например, Velero für Kubernetes).
### 5.4 Monitoring
```bash
# Docker Stats
docker stats nsct-api
# Container Logs (letzten 100 Zeilen)
docker compose logs --tail=100 nsct-api
# DB-Größe
docker exec -it nsct-postgres psql -U nsct -d nsct -c "SELECT pg_database_size('nsct');"
```
## 6. Troubleshooting
### Problem: API startet nicht
```bash
# Logs prüfen
docker compose logs nsct-api
# Häufige Ursachen:
# 1. .env-Datei nicht vorhanden
# 2. PostgreSQL nicht erreichbar
# 3. LLM-Endpoint nicht erreichbar
```
### Problem: PostgreSQL-Verbindung schlägt fehl
```bash
# Teste die Verbindung
docker compose exec postgres pg_isready -U nsct -d nsct
# Container-Logs prüfen
docker compose logs postgres
```
### Problem: LLM-Provider nicht erreichbar
```bash
# Network-Ping zum LLM-Host
docker compose run --rm nsct-api curl -v http://192.168.80.199:8030/openai/v1/models
```
## 7. Update / Migration
```bash
# Neueste Version holen
git pull
# Container rebuilden
docker compose up --build -d
# Datenbank-Migrationen (wenn benötigt)
# werden in späteren Stages mit Alembic implementiert
```

49
Dockerfile Normal file
View File

@@ -0,0 +1,49 @@
# ============================================================
# NSCT — Neutral Search Crawler Tool
# Multi-stage Docker build, non-root user
# ============================================================
# ---------- Build stage ----------
FROM python:3.12-slim AS builder
ENV PYTHONDONTWRITEBYTECODE=1 \
PYTHONUNBUFFERED=1 \
PIP_NO_CACHE_DIR=1 \
PIP_DEFAULT_TIMEOUT=60
RUN apt-get update && apt-get install -y --no-install-recommends \
gcc libpq-dev \
&& rm -rf /var/lib/apt/lists/*
WORKDIR /build
COPY pyproject.toml ./
RUN pip install --prefix=/install uvicorn fastapi pydantic pydantic-settings \
httpx sqlalchemy asyncpg beautifulsoup4 selectolax trafilatura structlog
COPY src/ ./src/
# ---------- Runtime stage ----------
FROM python:3.12-slim AS runtime
ENV PYTHONDONTWRITEBYTECODE=1 \
PYTHONUNBUFFERED=1
# Non-root user
RUN groupadd --gid 1000 nsct && \
useradd --uid 1000 --gid nsct --shell /bin/bash --create-home nsct
RUN mkdir -p /app/data && chown -R nsct:nsct /app/data
WORKDIR /app
COPY --from=builder /install /usr/local
COPY src/ ./src/
RUN chown -R nsct:nsct /app
USER nsct
EXPOSE 8080
CMD ["uvicorn", "nsct.api.main:app", "--host", "0.0.0.0", "--port", "8080"]

90
METHODOLOGY.md Normal file
View File

@@ -0,0 +1,90 @@
# NSCT — Methodik: Was "neutral" bedeutet
## 1. Grundverständnis von Neutralität
**Neutralität bedeutet nicht:** "Alle Meinungen sind gleich."
**Neutralität bedeutet nicht:** "Eine einfache Mehrheit gewinnt."
**Neutralität bedeutet:** "Unsicherheit wird explizit gemacht,
Quelle wird immer genannt, und der Bericht spiegelt die
tatsächliche Evidenzlage wider — ohne Bewertung."
## 2. Quellenabhängigkeit
NSCT ist **vollständig abhängig von der Qualität und Vielfalt
der verfügbaren Quellen**. Wenn eine Perspektive in den Quellen
fehlt, erscheint sie als "keine Quellen gefunden" — nicht als
"widerlegt".
Das bedeutet:
- **Keine Quellen → Unklarheit**, nicht falsche Aussage
- **Wenige Quellen → Niedrige Konfidenz**, nicht niedrige Wahrheit
- **Viele widersprüchliche Quellen → Disagreements**, nicht "irgendwer hat recht"
## 3. Claim-Vergleich statt Stimmzählung
NSCT durchsucht nicht nach "Anzahl der Quellen für X" und
ergibt dann "X ist wahr". Stattdessen:
1. **Extrahieren:** Alle Claims werden aus allen Quellen extrahiert
2. **Normalisieren:** Behauptungen werden in eine neutrale
Grundform gebracht (keine wertende Sprache)
3. **Vergleichen:** Claims werden paarweise verglichen
4. **Kategorisieren:**
- `agrees`: Direkte Übereinstimmung
- `disagrees`: Direkter Widerspruch
- `partially_agrees`: Teilweise Übereinstimmung
- `neutral`: Keine direkte Relation
- `contradicts`: Starker Widerspruch (stärker als disagreement)
5. **Berichten:** Alle Relationen werden dokumentiert mit Begründung
**Niemals** wird die Anzahl der zugunsten einer Aussage stehenden
Quellen als "Truth-Score" verwendet.
## 4. Unsicherheit als gültiges Resultat
Ein Bericht, der sagt "Es gibt keine ausreichende Evidenz für eine
klare Aussage zu diesem Punkt" ist ein **vollständiges und gültiges**
Ergebnis.
Unsicherheitskategorien:
- `low_confidence`: Nur wenige oder niedrig-qalifizierte Quellen
- `contradictory`: Quellen widersprechen sich deutlich
- `missing_perspective`: Eine plausible Perspektive wird nicht
durch Quellen vertreten
- `temporal`: Quellen sind veraltet oder aktuelle Entwicklungen
liegen nicht vor
## 5. Keine universelle "Truth Score"-Kennzahl
NSCT erzeugt **niemals** eine einzelne numerische Kennzahl wie
"Source Quality Score: 0.87" oder "Claim Truthiness: 92%".
Stattdessen:
- Jeder Claim hat eine extraktionsbezogene `confidence` (0-1)
- Jede EvidenceRelation hat eine `confidence` (0-1)
- Die Aggregation erfolgt durch Text (Summary, Findings,
Disagreements, Uncertainties), nicht durch Aggregation
numerischer Scores.
## 6. Quellen-Abhängigkeit vs. Quellen-Vertrauen
Ein wichtiger Unterschied:
- **Quellen-Abhängigkeit** ist eine technische Eigenschaft: "Welche
Quellen liegen vor und was sagen sie?"
- **Quellen-Vertrauen** ist eine bewertende Eigenschaft: "Wie sehr
darf man dieser Quelle glauben?"
NSCT macht **keine** Quellen-Vertrauen-Bewertung. Es dokumentiert
nur die Quellen-Abhängigkeit und lässt die Bewertung dem Nutzer.
## 7. Transparenz der Methodik
Jeder Bericht enthält ein `methodology`-Feld, das beschreibt:
- Wie viele Quellen durchsucht wurden
- Welche Search-Provider verwendet wurden
- Welche Claims extrahiert wurden
- Welche Relations zwischen Claims gefunden wurden
- Welche Unsicherheiten identifiziert wurden
Der Nutzer kann damit jederzeit nachvollziehen, wie der Bericht
zustande kam — und alternative Ansätze ausprobieren.

99
README.md Normal file
View File

@@ -0,0 +1,99 @@
# ============================================================
# NSCT — Neutral Search Crawler Tool
# ============================================================
**NSCT** ist ein vollständig lokal betreibbares, containerisiertes Recherche- und
Analyse-System. Es durchsucht Webquellen, extrahiert Inhalte, vergleicht
Behauptungen (Claims) aus verschiedenen Quellen und erzeugt einen neutralen,
quellengestützten Bericht.
**NSCT steht für:** Neutral Search Crawler Tool.
## Architektur-Übersicht
```
┌─────────────┐ ┌──────────────┐ ┌───────────────┐
│ User / CLI │───▶│ NSCT API │───▶│ FastAPI / │
│ │ │ (FastAPI) │ │ uvicorn │
└─────────────┘ └──────────────┘ └───────────────┘
│
┌────────────┼─────────────┐
▼ ▼ ▼
┌───────────┐ ┌─────────┐ ┌──────────┐
│ LLM │ │ Vision │ │ Audio │
│ Provider │ │ Model │ │ Model │
└───────────┘ └─────────┘ └──────────┘
│ │ │
▼ ▼ ▼
┌─────────────────────────────────────┐
│ PostgreSQL (evidence store) │
└─────────────────────────────────────┘
▲
│
┌──────────────┐
│ SearXNG │ (optional search backend)
└──────────────┘
```
## Quick Start
### Docker Compose
```bash
# 1. Kopiere die Beispiel-Env
cp .env.example .env
# 2. Trage deine Endpunkte ein (LLM, Vision, Audio, PostgreSQL)
# 3. Starte alles
docker compose up --build
# 4. Prüfe den Health-Check
curl http://localhost:8080/health
```
Die API ist danach unter `http://localhost:8080` erreichbar.
`/docs` zeigt die auto-generierte Swagger-Dokumentation (nur bei `NSCT_DEBUG=true`).
## Projektstruktur
```
nsct/
├── src/nsct/
│ ├── api/ # FastAPI-Routen
│ ├── models/ # Pydantic-Schemata
│ ├── providers/ # Abstrakte Provider-Interfaces
│ ├── security/ # SSRF-Schutz, URL-Validierung
│ └── storage/ # SQLAlchemy Models + Engine
├── tests/ # pytest-Tests
├── docker-compose.yml
├── Dockerfile
├── pyproject.toml
├── README.md
├── ARCHITECTURE.md
├── SECURITY.md
├── METHODOLOGY.md
├── API.md
├── DEPLOYMENT.md
└── .env.example
```
## Status
**Stage 0** — Repository und Architekturgrundlage.
Stage 0 enthält:
- Vollständige Projektstruktur mit allen Konfigurationsdateien
- Pydantic v2 Datenmodelle
- SQLAlchemy 2.0 Persistenzmodelle
- Sicherheitshards (SSRF-Schutz, IP-Blocklist)
- Provider-Interfaces (abstrakte Basisklassen)
- Health-, Ready- und Provider-Endpunkte
- Strukturiertes Logging
- Docker-Konfiguration für PostgreSQL, SearXNG und NSCT
Die eigentliche Evidence-Pipeline (Search, Fetch, Extract, Claim, Compare, Report)
wird in späteren Stages implementiert.
## Lizenz
MIT

105
SECURITY.md Normal file
View File

@@ -0,0 +1,105 @@
# NSCT — Sicherheitsgrundsätze
## 1. Grundprinzip
NSCT verarbeitet **untrusted input** von überall im Web. Jede externe Quelle
kann versuchen, die Integrität des Systems zu kompromittieren. Alle
Sicherheitsmaßnahmen folgen dem "zero trust" Prinzip.
## 2. Control Plane vs. Evidence Plane
```
┌─────────────────────────────────────────────────────────┐
│ CONTROL PLANE │
│ (vertrauenswürdige, interne Daten) │
│ │
│ • Konfiguration (Environment Variables) │
│ • Strukturierter Prompt (LLM Instruktionen) │
│ • Normierte Claims (bereinigt, de-biased) │
│ • EvidenceRelations (Vergleichsergebnisse) │
└──────────────┬──────────────────────────────────────────┘
│
│ LLM verarbeitet NUR strukturierte Daten
│
┌──────────────▼──────────────────────────────────────────┐
│ EVIDENCE PLANE │
│ (untrusted, externe Daten) │
│ │
│ • Roh-Webcontent (HTML, Text) │
│ • Search-Provider-Ergebnisse │
│ • Extrahierte Claims (roh, unverarbeitet) │
│ • Audio/Vision-Rohdaten │
└─────────────────────────────────────────────────────────┘
```
**Regel:** Webcontent wechselt niemals direkt die Grenze von Evidence Plane
in die Control Plane als Prompt-Text.
## 3. SSRF-Schutz
Jede outbound-HTTP-Anfrage muss durch `nsct/security/policy.py`:
1. **Schema-Check:** Nur `http://` und `https://` erlaubt
2. **IP-Blocklist:**
- `127.0.0.0/8` — Loopback
- `10.0.0.0/8` — Private Class A
- `172.16.0.0/12` — Private Class B
- `192.168.0.0/16` — Private Class C
- `169.254.0.0/16` — Link-local / Link-local
- `192.0.0.0/24` — IETF Protocol Assignments
- `100.64.0.0/10` — Shared Address Space
- `192.0.2.0/24`, `198.51.100.0/24`, `203.0.113.0/24` — TEST-NET
3. **Metadata-Endpunkte:** Cloud-Instance-Metadata (`169.254.169.254`, etc.)
4. **Schemata:** `file://`, `ftp://`, `gopher://`, `ldap://`, `data://` blockiert
## 4. Prompt-Isolation
Der LLM darf **niemals** rohen Webcontent als Teil des Prompts sehen. Der
einzige Weg, Webcontent für den LLM verfügbar zu machen, ist über strukturierte
Daten:
```json
{
"source": {
"url": "https://example.com",
"domain": "example.com",
"title": "Titel",
"author": "Autor",
"content_type": "news",
"language": "de"
},
"claims": [
{
"id": "uuid",
"normalized_claim": "De-biased claim text",
"claim_type": "factual",
"confidence": 0.95
}
]
}
```
Der LLM sieht **nur** dieses JSON — nie den HTML/Raw-Text.
## 5. Untrusted Data Handling
| Quelle | Risiko | Schutzmaßnahme |
|--------|--------|---------------|
| Search Provider | Fake-Scores, Manipulation | Score != Evidence-Ranking; nur als Eingabe |
| Web Content | XSS, Prompt-Injection, SSRF | HTML-Extraktion → Text, strikte Schema-Validierung |
| LLM Output | Halluzination, Bias | Claims mit confidence scoring; keine single-source truth |
| Audio/Vision | Manipulierte Inputs | Content-Type-Validierung; Größ Limits |
## 6. Docker-Sicherheit
- **Non-Root-User:** `nsct` (UID 1000)
- **Read-Only-Filesystem:** `read_only: true` wo möglich
- **Dropped Capabilities:** `cap_drop: [ALL]`
- **No Secrets in Image:** `.env` wird nicht gebuildet
- **Resource Limits:** Memory und CPU begrenzt pro Container
## 7. Compliance
- Keine Speicherung von personenbezogenen Daten über die Quellen-Metadaten hinaus
- Claims und Evidence Relations werden nur so lange gespeichert wie nötig
- Alle Daten sind exportierbar und löschbar (SQLAlchemy cascade)

105
docker-compose.yml Normal file
View File

@@ -0,0 +1,105 @@
# ============================================================
# NSCT — docker-compose.yml
# Services: nsct-api, postgres, optional searxng
# ============================================================
x-common: &common
env_file:
- .env
restart: unless-stopped
deploy:
resources:
limits:
memory: 1G
cpus: "1.0"
reservations:
memory: 256M
cpus: "0.25"
services:
# ---------- PostgreSQL ----------
postgres:
image: postgres:16-alpine
container_name: nsct-postgres
environment:
POSTGRES_USER: ${POSTGRES_USER:-nsct}
POSTGRES_PASSWORD: ${POSTGRES_PASSWORD:-nsct_secret}
POSTGRES_DB: ${POSTGRES_DB:-nsct}
ports:
- "5432:5432"
volumes:
- postgres_data:/var/lib/postgresql/data
healthcheck:
test: ["CMD-SHELL", "pg_isready -U ${POSTGRES_USER:-nsct} -d ${POSTGRES_DB:-nsct}"]
interval: 10s
timeout: 5s
retries: 5
start_period: 10s
restart: unless-stopped
deploy:
resources:
limits:
memory: 512M
cpus: "0.5"
# ---------- SearXNG (optional) ----------
searxng:
image: searxng/searxng:latest
container_name: nsct-searxng
ports:
- "8888:8080"
volumes:
- searxng_settings:/etc/searxng
environment:
- SEARXNG_BASE_URL=http://localhost:8888/
- SEARXNG_SECRET_KEY=nsct_searxng_secret_key_change_me
restart: unless-stopped
deploy:
resources:
limits:
memory: 512M
cpus: "0.5"
read_only: true
tmpfs:
- /tmp
cap_drop:
- ALL
# ---------- NSCT API ----------
nsct-api:
build:
context: .
dockerfile: Dockerfile
container_name: nsct-api
<<: *common
ports:
- "8080:8080"
environment:
NSCT_DB_URL: postgresql+asyncpg://${POSTGRES_USER:-nsct}:${POSTGRES_PASSWORD:-nsct_secret}@postgres:5432/${POSTGRES_DB:-nsct}
NSCT_LLM_MAX_CONCURRENCY: "${NSCT_LLM_MAX_CONCURRENCY:-3}"
depends_on:
postgres:
condition: service_healthy
searxng:
condition: service_started
volumes:
- nsct_data:/app/data
healthcheck:
test: ["CMD-SHELL", "curl -f http://localhost:8080/health || exit 1"]
interval: 30s
timeout: 5s
retries: 3
start_period: 15s
read_only: true
tmpfs:
- /tmp
cap_drop:
- ALL
volumes:
postgres_data:
driver: local
nsct_data:
driver: local
searxng_settings:
driver: local

1871
prompt.md Normal file
View File

File diff suppressed because it is too large Load Diff

56
pyproject.toml Normal file
View File

@@ -0,0 +1,56 @@
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[project]
name = "nsct"
version = "0.1.0"
description = "Neutral Search Crawler Tool — lokal betreibbares Recherche- und Analyse-System"
readme = "README.md"
license = "MIT"
requires-python = ">=3.12"
dependencies = [
"fastapi>=0.115.0",
"pydantic>=2.0,<3.0",
"pydantic-settings>=2.0,<3.0",
"httpx>=0.27.0",
"sqlalchemy>=2.0,<3.0",
"asyncpg>=0.30.0",
"beautifulsoup4>=4.12.0",
"selectolax>=0.3.0",
"trafilatura>=2.0.0",
"uvicorn>=0.30.0",
"structlog>=24.0.0",
]
[project.optional-dependencies]
dev = [
"pytest>=8.0.0",
"pytest-asyncio>=0.24.0",
"pytest-cov>=5.0.0",
"aiosqlite>=0.20.0",
"coverage>=7.0.0",
]
postgresql = [
"asyncpg>=0.30.0",
]
[project.scripts]
nsct = "nsct.cli:main"
nsct-core = "nsct.cli:main"
nsct-api = "nsct.cli:main_api"
[tool.hatch.build]
packages = ["src/nsct"]
[tool.hatch.build.targets.wheel]
packages = ["nsct"]
[tool.pytest.ini_options]
asyncio_mode = "auto"
testpaths = ["tests"]
pythonpath = ["src"]
[tool.ruff]
target-version = "py312"
line-length = 120

3
src/nsct/__init__.py Normal file
View File

@@ -0,0 +1,3 @@
"""NSCT — Neutral Search Crawler Tool."""
__version__ = "0.1.0"

1
src/nsct/api/__init__.py Normal file
View File

@@ -0,0 +1 @@
"""NSCT — package init."""

84
src/nsct/api/health.py Normal file
View File

@@ -0,0 +1,84 @@
"""Health check router for the FastAPI application."""
from __future__ import annotations
import httpx
import logging
from fastapi import APIRouter
from nsct.config import AppSettings
logger = logging.getLogger(__name__)
router = APIRouter()
@router.get("/health")
def health_check():
"""Basic health check — always returns ok when the process is alive."""
from nsct import __version__
return {
"status": "ok",
"version": __version__,
}
@router.get("/ready")
async def readiness_check(config: AppSettings | None = None):
"""Readiness check — probes downstream services."""
result: dict[str, str | bool] = {
"status": "not_ready",
"llm": "error",
"database": "unknown",
}
# LLM connectivity check
llm_base = config.llm.base_url if config and config.llm and config.llm.base_url else ""
if llm_base:
try:
async with httpx.AsyncClient(timeout=3.0) as client:
resp = await client.get(f"{llm_base}/models")
if resp.status_code == 200:
result["llm"] = "ok"
result["llm_model"] = config.llm.model if config else "unknown"
else:
result["llm"] = f"error:{resp.status_code}"
except Exception as exc:
result["llm"] = f"error:{type(exc).__name__}"
if result.get("llm") == "ok":
result["status"] = "ready"
return result
@router.get("/providers")
def providers_info(config: AppSettings | None = None):
"""List available providers — NO secrets returned."""
if not config:
return {
"llm": {"available": False},
"vision": {"available": False},
"audio": {"available": False},
}
llm_available = bool(config.llm.base_url)
vision_available = bool(config.vision.base_url)
audio_available = bool(config.audio.base_url)
return {
"llm": {
"available": llm_available,
"model": config.llm.model if llm_available else None,
"max_concurrency": config.llm.max_concurrency if llm_available else None,
},
"vision": {
"available": vision_available,
"model": config.vision.model if vision_available else None,
},
"audio": {
"available": audio_available,
"model": config.audio.model if audio_available else None,
},
}

81
src/nsct/api/main.py Normal file
View File

@@ -0,0 +1,81 @@
"""FastAPI application — entry point for the API server."""
from __future__ import annotations
import logging
from contextlib import asynccontextmanager
import httpx
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from nsct.config import AppSettings
logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# Lifespan
# ---------------------------------------------------------------------------
@asynccontextmanager
async def lifespan(app: FastAPI):
"""Application lifespan — startup and shutdown hooks."""
config: AppSettings = app.state.config
logger.info("NSCT API starting up — version %s", app.state.version)
# Pre-flight: validate LLM connectivity
llm_base = config.llm.base_url if config and config.llm and config.llm.base_url else ""
if llm_base:
try:
async with httpx.AsyncClient(timeout=5.0) as client:
resp = await client.get(f"{llm_base}/models")
if resp.status_code == 200:
logger.info("LLM provider reachable — model endpoint returned 200")
else:
logger.warning("LLM provider returned %s", resp.status_code)
except Exception as exc:
logger.warning("LLM connectivity check failed: %s", exc)
yield
# ---------------------------------------------------------------------------
# App factory
# ---------------------------------------------------------------------------
def create_app() -> FastAPI:
"""Create the FastAPI application with all routes and middleware."""
config = AppSettings.from_env()
app = FastAPI(
title="NSCT API",
description="Neutral Search Crawler Tool — API server",
version="0.1.0",
docs_url="/docs" if config.debug else None,
redoc_url="/redoc" if config.debug else None,
lifespan=lifespan,
)
app.state.config = config
app.state.version = "0.1.0"
# CORS — allow local development
app.add_middleware(
CORSMiddleware,
allow_origins=["http://localhost:3000", "http://localhost:5173", "http://127.0.0.1:3000"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Mount health router
from nsct.api.health import router as health_router
app.include_router(health_router, tags=["system"])
return app
app = create_app()

65
src/nsct/cli.py Normal file
View File

@@ -0,0 +1,65 @@
"""CLI entry-point for NSCT — stub for future CLI commands."""
from __future__ import annotations
import argparse
import sys
from nsct import __version__
def main() -> int:
"""Main CLI entry-point."""
parser = argparse.ArgumentParser(
prog="nsct",
description="NSCT — Neutral Search Crawler Tool",
)
parser.add_argument("--version", action="version", version=f"NSCT v{__version__}")
parser.add_argument(
"command",
nargs="?",
default="help",
help="Command to run (help, search, report, etc.)",
)
args = parser.parse_args()
if args.command == "help":
print("NSCT — Neutral Search Crawler Tool")
print(f"Version: {__version__}")
print()
print("Commands:")
print(" help Show this help message")
print(" search Search web sources (coming in Stage 1)")
print(" report Generate a research report (coming in Stage 2)")
print(" version Show version")
return 0
print(f"Unknown command: {args.command}")
return 1
def main_api() -> int:
"""Start the FastAPI server directly from CLI."""
import uvicorn
from nsct.config import AppSettings
config = AppSettings.from_env()
host = "0.0.0.0"
port = 8080
print(f"Starting NSCT API server on {host}:{port}")
print(f"LLM model: {config.llm.model}")
print(f"Debug mode: {config.debug}")
uvicorn.run(
"nsct.api.main:app",
host=host,
port=port,
log_level="info" if not config.debug else "debug",
)
return 0
if __name__ == "__main__":
sys.exit(main())

116
src/nsct/config.py Normal file
View File

@@ -0,0 +1,116 @@
"""Central configuration — all values from environment, zero hard-coded secrets."""
from __future__ import annotations
import os
from pathlib import Path
from typing import Literal
from pydantic import BaseModel, Field, field_validator
class LLMConfig(BaseModel):
"""LLM worker configuration."""
base_url: str = Field(..., description="Base URL of the LLM provider (OpenAI-compatible).")
model: str = Field(..., description="Model identifier to use for LLM calls.")
max_concurrency: int = Field(default=3, ge=1, description="Max parallel LLM requests.")
def get_secret(self) -> str:
"""Return the raw API key string if present."""
key = os.environ.get("NSCT_LLM_API_KEY", "")
return key if key else ""
class VisionConfig(BaseModel):
"""Vision (image/video) model configuration."""
base_url: str = Field(..., description="Base URL of the vision provider endpoint.")
model: str = Field(..., description="Vision model identifier.")
def get_secret(self) -> str:
key = os.environ.get("NSCT_VISION_API_KEY", "")
return key if key else ""
class AudioConfig(BaseModel):
"""Audio / speech model configuration."""
base_url: str = Field(..., description="Base URL of the audio provider endpoint.")
model: str = Field(default="default", description="Audio model identifier.")
def get_secret(self) -> str:
key = os.environ.get("NSCT_AUDIO_API_KEY", "")
return key if key else ""
class DatabaseConfig(BaseModel):
"""Database connection configuration."""
url: str = Field(
...,
description="Database URL (postgresql+asyncpg://...) or a default SQLite fallback.",
)
class AppSettings(BaseModel):
"""Top-level application settings."""
llm: LLMConfig = Field(default_factory=LLMConfig)
vision: VisionConfig = Field(default_factory=VisionConfig)
audio: AudioConfig = Field(default_factory=AudioConfig)
postgres: DatabaseConfig = Field(default_factory=DatabaseConfig)
debug: bool = Field(default=False)
searxng_base_url: str | None = Field(default=None, description="SearXNG instance URL.")
@staticmethod
def from_env() -> "AppSettings":
"""Build AppSettings entirely from environment variables."""
llm_base_url = os.environ.get("NSCT_LLM_BASE_URL", "")
llm_model = os.environ.get("NSCT_LLM_MODEL", "")
llm_concurrency = int(os.environ.get("NSCT_LLM_MAX_CONCURRENCY", "3"))
vision_base_url = os.environ.get("NSCT_VISION_BASE_URL", "")
vision_model = os.environ.get("NSCT_VISION_MODEL", "")
audio_base_url = os.environ.get("NSCT_AUDIO_BASE_URL", "")
audio_model = os.environ.get("NSCT_AUDIO_MODEL", "default")
db_url = os.environ.get("NSCT_DB_URL", "")
llm_cfg = LLMConfig(
base_url=llm_base_url,
model=llm_model,
max_concurrency=llm_concurrency,
)
vision_cfg = VisionConfig(
base_url=vision_base_url,
model=vision_model,
)
audio_cfg = AudioConfig(
base_url=audio_base_url,
model=audio_model,
)
postgres_cfg = DatabaseConfig(url=db_url)
debug = os.environ.get("NSCT_DEBUG", "false").lower() == "true"
searxng_url = os.environ.get("NSCT_SEARXNG_BASE_URL", None)
return AppSettings(
llm=llm_cfg,
vision=vision_cfg,
audio=audio_cfg,
postgres=postgres_cfg,
debug=debug,
searxng_base_url=searxng_url,
)
@property
def config_dir(self) -> Path:
"""Return a config directory path for persisting runtime data."""
data_dir = Path.home() / ".nsct"
data_dir.mkdir(parents=True, exist_ok=True)
return data_dir

126
src/nsct/logging_config.py Normal file
View File

@@ -0,0 +1,126 @@
"""Structured logging with research_id and llm_request_id tracking.
Uses the standard library logging module with a JSON formatter so that every
log record is machine-parseable and traceable across distributed components.
"""
from __future__ import annotations
import json
import logging
import sys
import time
import uuid
from typing import Any
from nsct import __version__
# ---------------------------------------------------------------------------
# Global request-context helpers — all structured log calls pick them up.
# ---------------------------------------------------------------------------
_request_ctx: dict[str, str] = {}
def set_request_ctx(**kwargs: str) -> None:
"""Set (merge) key-value pairs into the per-request context dict."""
_request_ctx.update(kwargs)
def clear_request_ctx() -> None:
"""Clear all per-request context entries."""
_request_ctx.clear()
# ---------------------------------------------------------------------------
# JSON Formatter
# ---------------------------------------------------------------------------
class JSONFormatter(logging.Formatter):
"""Emit log records as a single JSON object per line."""
def __init__(self, *args: Any, **kwargs: Any) -> None:
super().__init__(*args, **kwargs)
self._start_ts = time.monotonic()
# no default format — we build the dict ourselves
default_fmt: str = ""
def format(self, record: logging.LogRecord) -> str:
ts = time.strftime("%Y-%m-%dT%H:%M:%S", time.gmtime(record.created))
elapsed_ms = round((time.monotonic() - self._start_ts) * 1000, 2)
extra = {
"_ts": ts,
"_ms": elapsed_ms,
"_pid": record.process,
"_thread": record.thread,
"_level": record.levelname,
"_module": record.module,
"_function": record.funcName,
"_line": record.lineno,
"_version": __version__,
}
# pull in the request context
for key, val in _request_ctx.items():
extra[f"ctx.{key}"] = val
# merge with standard fields
data: dict[str, Any] = dict(extra)
data["msg"] = record.getMessage()
data["name"] = record.name
# exception info
if record.exc_info and record.exc_info[0] is not None:
data["exc_info"] = self.formatException(record.exc_info)
return json.dumps(data, default=str, ensure_ascii=False)
# ---------------------------------------------------------------------------
# Bootstrap
# ---------------------------------------------------------------------------
def setup_logging(
level: str = "INFO",
*,
json_format: bool = True,
) -> None:
"""Configure the root logger with a single console handler.
Args:
level: Log level string (DEBUG, INFO, WARNING, ERROR, CRITICAL).
json_format: If True, use JSONFormatter; else use a human-friendly format.
"""
log_level = getattr(logging, level.upper(), logging.INFO)
root = logging.getLogger()
root.setLevel(log_level)
handler = logging.StreamHandler(sys.stdout)
handler.setLevel(log_level)
if json_format:
handler.setFormatter(JSONFormatter())
else:
handler.setFormatter(
logging.Formatter(
"%(asctime)s [%(levelname)-8s] %(name)s:%(funcName)s:%(lineno)d — %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
)
)
root.addHandler(handler)
# ---------------------------------------------------------------------------
# Convenience logger factory
# ---------------------------------------------------------------------------
def get_logger(name: str = __name__) -> logging.Logger:
"""Return a pre-configured child logger."""
return logging.getLogger(name)

184
src/nsct/models/schemas.py Normal file
View File

@@ -0,0 +1,184 @@
"""Pydantic v2 schemas for NSCT data objects."""
from __future__ import annotations
from datetime import datetime
from enum import Enum
from typing import Any
from uuid import UUID, uuid4
from pydantic import BaseModel, Field
# ---------------------------------------------------------------------------
# Enums
# ---------------------------------------------------------------------------
class ClaimType(str, Enum):
"""Types of claims that can be extracted from sources."""
FACTUAL = "factual"
OPINION = "opinion"
PREDICTION = "prediction"
EVALUATION = "evaluation"
COMPARISON = "comparison"
class EvidenceRelationType(str, Enum):
"""Relationship types between two claims."""
AGREES = "agrees"
DISAGREES = "disagrees"
NEUTRAL = "neutral"
PARTIALLY_AGREES = "partially_agrees"
PARTIALLY_DISAGREES = "partially_disagrees"
CONTRADICTS = "contradicts"
class SourceType(str, Enum):
"""Classification of a source."""
NEWS = "news"
ACADEMIC = "academic"
BLOG = "blog"
GOVERNMENT = "government"
CORPORATE = "corporate"
SOCIAL_MEDIA = "social_media"
DOCUMENT = "document"
OTHER = "other"
class EdgeRelation(str, Enum):
"""Relationship between two sources."""
CITATION = "citation"
CORROBORATION = "corroboration"
CONTRADICTION = "contradiction"
DEPENDS_ON = "depends_on"
SUPPLEMENTS = "supplements"
# ---------------------------------------------------------------------------
# SearchQuery
# ---------------------------------------------------------------------------
class SearchQuery(BaseModel):
"""Represents a user-initiated search/research query."""
id: UUID = Field(default_factory=uuid4)
research_id: UUID = Field(default_factory=uuid4)
query: str = Field(..., min_length=1, description="The search query text.")
purpose: str | None = Field(default=None, description="Intent behind the query.")
language: str = Field(default="de", description="Language code, e.g. 'de', 'en'.")
category: str | None = Field(default=None, description="Optional category for grouping.")
created_at: datetime = Field(default_factory=datetime.utcnow)
model_config = {"frozen": True}
# ---------------------------------------------------------------------------
# Source
# ---------------------------------------------------------------------------
class Source(BaseModel):
"""A single retrieved source (web page, document, etc.)."""
id: UUID = Field(default_factory=uuid4)
url: str = Field(..., description="Original URL.")
canonical_url: str | None = Field(default=None, description="Canonical URL after redirect resolution.")
domain: str = Field(..., description="Extracted domain.")
title: str | None = Field(default=None)
author: str | None = Field(default=None)
publisher: str | None = Field(default=None)
publication_date: datetime | None = Field(default=None)
retrieved_at: datetime = Field(default_factory=datetime.utcnow)
content_type: str | None = Field(default=None, description="MIME type.")
source_type: SourceType | None = Field(default=None)
language: str = Field(default="unknown")
content_hash: str | None = Field(default=None, description="SHA-256 hash of raw content.")
parent_source_id: UUID | None = Field(default=None, description="Parent source for mirrors/canonical pairs.")
content: str | None = Field(default=None, description="Extracted text content.")
model_config = {"frozen": True}
# ---------------------------------------------------------------------------
# Claim
# ---------------------------------------------------------------------------
class Claim(BaseModel):
"""A factual assertion extracted from a Source."""
id: UUID = Field(default_factory=uuid4)
source_id: UUID = Field(..., description="UUID of the Source this claim came from.")
claim: str = Field(..., description="Original claim text.")
normalized_claim: str | None = Field(default=None, description="De-biased / neutral claim wording.")
claim_type: ClaimType = Field(default=ClaimType.FACTUAL)
subject: str | None = Field(default=None)
predicate: str | None = Field(default=None)
object: str | None = Field(default=None)
evidence_span: str | None = Field(default=None, description="Span of text in the source that supports this claim.")
confidence: float = Field(default=1.0, ge=0.0, le=1.0, description="Extraction confidence (0-1).")
event_date: datetime | None = Field(default=None)
model_config = {"frozen": True}
# ---------------------------------------------------------------------------
# EvidenceRelation
# ---------------------------------------------------------------------------
class EvidenceRelation(BaseModel):
"""Relationship between two claims."""
claim_a: UUID = Field(..., description="UUID of the first claim.")
claim_b: UUID = Field(..., description="UUID of the second claim.")
relation: EvidenceRelationType = Field(...)
confidence: float = Field(default=0.5, ge=0.0, le=1.0)
reason: str | None = Field(default=None, description="Free-text explanation of the relation.")
# ---------------------------------------------------------------------------
# CitationEdge
# ---------------------------------------------------------------------------
class CitationEdge(BaseModel):
"""Directed edge between two sources indicating a relationship."""
source_from: UUID = Field(..., description="UUID of the citing source.")
source_to: UUID = Field(..., description="UUID of the cited source.")
relation: EdgeRelation = Field(...)
confidence: float = Field(default=1.0, ge=0.0, le=1.0)
# ---------------------------------------------------------------------------
# ResearchReport
# ---------------------------------------------------------------------------
class ResearchReport(BaseModel):
"""Final aggregated research report."""
research_id: UUID = Field(default_factory=uuid4)
query: str = Field(...)
summary: str = Field(default="", description="Human-readable summary.")
findings: list[str] = Field(default_factory=list, description="List of key findings.")
disagreements: list[dict[str, Any]] = Field(
default_factory=list, description="Conflicting claims with details."
)
uncertainties: list[str] = Field(default_factory=list, description="Known uncertainties.")
source_statistics: dict[str, Any] = Field(
default_factory=dict, description="Stats: counts per source type, language, etc."
)
methodology: str = Field(
default="", description="Description of the methodology used to produce the report."
)
generated_at: datetime = Field(default_factory=datetime.utcnow)
model_config = {"frozen": True}

1
src/nsct/providers/__init__.py Normal file
View File

@@ -0,0 +1 @@
"""NSCT — providers package init."""

1
src/nsct/security/__init__.py Normal file
View File

@@ -0,0 +1 @@
"""NSCT — security package init."""

139
src/nsct/security/policy.py Normal file
View File

@@ -0,0 +1,139 @@
"""Security policy — SSRF protection, URL validation, and IP blocklisting.
All outbound HTTP requests from NSCT MUST pass through this module.
"""
from __future__ import annotations
import ipaddress
import logging
import re
from urllib.parse import urlparse
logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# Blocked address ranges
# ---------------------------------------------------------------------------
_BLOCKED_NETWORKS: list[ipaddress.IPv4Network] = [
ipaddress.IPv4Network("127.0.0.0/8"), # Loopback
ipaddress.IPv4Network("10.0.0.0/8"), # Private
ipaddress.IPv4Network("172.16.0.0/12"), # Private
ipaddress.IPv4Network("192.168.0.0/16"), # Private
ipaddress.IPv4Network("169.254.0.0/16"), # Link-local
ipaddress.IPv4Network("0.0.0.0/8"), # "This" network
ipaddress.IPv4Network("100.64.0.0/10"), # Shared (RFC 6598)
ipaddress.IPv4Network("192.0.0.0/24"), # IETF protocol assignments
ipaddress.IPv4Network("192.0.2.0/24"), # TEST-NET-1
ipaddress.IPv4Network("198.51.100.0/24"), # TEST-NET-2
ipaddress.IPv4Network("203.0.113.0/24"), # TEST-NET-3
]
# IPv6 equivalents (loopback, unique-local, link-local)
_BLOCKED_IPV6_NETWORKS: list[ipaddress.IPv6Network] = [
ipaddress.IPv6Network("::1/128"), # Loopback
ipaddress.IPv6Network("::1/128"),
ipaddress.IPv6Network("fc00::/7"), # Unique-local
ipaddress.IPv6Network("fe80::/10"), # Link-local
]
# Blocked schemes
_BLOCKED_SCHEMES = frozenset(["file", "ftp", "gopher", "ldap", "mailto", "data", "blob"])
# Cloud metadata endpoint pattern
_METADATA_PATTERNS = [
"169.254.169.254",
"metadata.google.internal",
"metadata.aws.internal",
"169.254.170.2",
"169.254.169.254",
"instance-data",
"amazonaws.com",
]
# ---------------------------------------------------------------------------
# Validation
# ---------------------------------------------------------------------------
class SSRFError(Exception):
"""Raised when a URL is deemed unsafe for outbound access."""
class SSRFValidationError(SSRFError):
"""Specifically raised for SSRF-related validation failures."""
class InvalidURLError(SSRFError):
"""Raised when the URL is structurally invalid."""
def validate_url(url: str) -> None:
"""Validate a URL against SSRF rules.
Raises SSRFError / InvalidURLError on violation.
"""
if not url or not isinstance(url, str):
raise InvalidURLError("URL must be a non-empty string")
# --- Scheme check ---
parsed = urlparse(url)
scheme = parsed.scheme.lower()
if scheme not in ("http", "https"):
raise InvalidURLError(f"Scheme '{scheme}' is not allowed (only http/https)")
# --- Blocked schemes ---
if scheme in _BLOCKED_SCHEMES:
raise InvalidURLError(f"Scheme '{scheme}' is explicitly blocked")
# --- Host check ---
host = parsed.hostname or ""
if not host:
raise InvalidURLError("URL has no hostname")
# --- Check for localhost / private IPs ---
try:
addr = ipaddress.ip_address(host)
except ValueError:
# Hostname — do DNS-based resolution (caller must handle)
pass
else:
# IPv4 blocked networks
if isinstance(addr, ipaddress.IPv4Address):
for net in _BLOCKED_NETWORKS:
if addr in net:
raise SSRFValidationError(
f"IP {addr} is in a blocked range"
)
# IPv6 blocked networks
if isinstance(addr, ipaddress.IPv6Address):
for net in _BLOCKED_IPV6_NETWORKS:
if addr in net:
raise SSRFValidationError(
f"IPv6 address {addr} is in a blocked range"
)
# --- Cloud metadata detection ---
host_lower = host.lower()
for pattern in _METADATA_PATTERNS:
if pattern in host_lower:
raise SSRFValidationError(
f"Hostname '{host}' matches cloud metadata pattern"
)
logger.info("URL validated: %s", url)
def is_safe_host(host: str) -> bool:
"""Quick hostname check — returns False for obviously unsafe hosts.
This is a first-pass filter; callers should still call validate_url().
"""
lower = host.lower()
for pattern in _METADATA_PATTERNS:
if pattern in lower:
return False
return True

1
src/nsct/storage/__init__.py Normal file
View File

@@ -0,0 +1 @@
"""NSCT — storage package init."""

83
src/nsct/storage/engine.py Normal file
View File

@@ -0,0 +1,83 @@
"""Async SQLAlchemy engine factory — creates and manages the connection pool."""
from __future__ import annotations
import logging
from typing import AsyncGenerator
from sqlalchemy import event
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from nsct.config import AppSettings
from nsct.storage.models import Base # noqa: F401 — Base is needed for create_all
logger = logging.getLogger(__name__)
_engine = None
_session_factory = None
async def get_engine(config: AppSettings) -> None:
"""Initialise the global async engine and session factory.
Must be called once during application startup (lifespan) before any
request is handled.
"""
global _engine, _session_factory
if _engine is not None:
logger.warning("Engine already initialised; skipping create_engine.")
return
db_url = config.postgres.url
_engine = create_async_engine(
db_url,
echo=config.debug,
pool_size=10,
max_overflow=20,
pool_recycle=1800,
pool_pre_ping=True,
)
# Ensure tables exist (schema migration not handled here — that is a
# separate migration step; this creates missing tables only).
async with _engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
_session_factory = async_sessionmaker(
_engine,
class_=AsyncSession,
expire_on_commit=False,
autocommit=False,
autoflush=False,
)
logger.info("Async SQLAlchemy engine initialised. DB: %s", db_url)
async def get_session() -> AsyncGenerator[AsyncSession, None]:
"""Yield an async session for request-scoped database access."""
global _session_factory
if _session_factory is None:
raise RuntimeError("Engine not initialised. Call get_engine() first.")
session = _session_factory()
try:
yield session
await session.commit()
except Exception:
await session.rollback()
raise
finally:
await session.close()
async def close_engine() -> None:
"""Close the engine and release all pooled connections."""
global _engine, _session_factory
if _engine is not None:
await _engine.dispose()
logger.info("SQLAlchemy engine disposed.")
_engine = None
_session_factory = None

210
src/nsct/storage/models.py Normal file
View File

@@ -0,0 +1,210 @@
"""SQLAlchemy 2.0 Declarative models — mapping of NSCT domain objects to tables."""
from __future__ import annotations
import enum
from datetime import datetime
from typing import Any
from uuid import UUID, uuid4
from sqlalchemy import (
BigInteger,
Column,
DateTime,
Enum,
Float,
ForeignKey,
Index,
Integer,
String,
Text,
)
from sqlalchemy.orm import DeclarativeBase, relationship
class ClaimType(str, enum.Enum):
FACTUAL = "factual"
OPINION = "opinion"
PREDICTION = "prediction"
EVALUATION = "evaluation"
COMPARISON = "comparison"
class SourceType(str, enum.Enum):
NEWS = "news"
ACADEMIC = "academic"
BLOG = "blog"
GOVERNMENT = "government"
CORPORATE = "corporate"
SOCIAL_MEDIA = "social_media"
DOCUMENT = "document"
OTHER = "other"
class EvidenceRelationType(str, enum.Enum):
AGREES = "agrees"
DISAGREES = "disagrees"
NEUTRAL = "neutral"
PARTIALLY_AGREES = "partially_agrees"
PARTIALLY_DISAGREES = "partially_disagrees"
CONTRADICTS = "contradicts"
class EdgeRelation(str, enum.Enum):
CITATION = "citation"
CORROBORATION = "corroboration"
CONTRADICTION = "contradiction"
DEPENDS_ON = "depends_on"
SUPPLEMENTS = "supplements"
# ---------------------------------------------------------------------------
# Base
# ---------------------------------------------------------------------------
class Base(DeclarativeBase):
pass
# ---------------------------------------------------------------------------
# SearchQuery
# ---------------------------------------------------------------------------
class SearchQueryModel(Base):
__tablename__ = "search_queries"
id = Column(String(36), primary_key=True, default=lambda: str(uuid4()))
research_id = Column(String(36), nullable=False, default=lambda: str(uuid4()))
query = Column(Text, nullable=False)
purpose = Column(Text, nullable=True)
language = Column(String(16), nullable=False, default="de")
category = Column(String(64), nullable=True)
created_at = Column(DateTime, nullable=False, default=datetime.utcnow)
__table_args__ = (Index("ix_search_queries_research_id", "research_id"),)
# ---------------------------------------------------------------------------
# Source
# ---------------------------------------------------------------------------
class SourceModel(Base):
__tablename__ = "sources"
id = Column(String(36), primary_key=True, default=lambda: str(uuid4()))
url = Column(Text, nullable=False)
canonical_url = Column(Text, nullable=True)
domain = Column(String(255), nullable=False)
title = Column(Text, nullable=True)
author = Column(Text, nullable=True)
publisher = Column(Text, nullable=True)
publication_date = Column(DateTime, nullable=True)
retrieved_at = Column(DateTime, nullable=False, default=datetime.utcnow)
content_type = Column(String(128), nullable=True)
source_type = Column(Enum(SourceType), nullable=True)
language = Column(String(16), nullable=False, default="unknown")
content_hash = Column(String(64), nullable=True)
parent_source_id = Column(String(36), ForeignKey("sources.id"), nullable=True)
content = Column(Text, nullable=True)
# Relationships
claims = relationship("ClaimModel", back_populates="source", cascade="all, delete-orphan")
__table_args__ = (
Index("ix_sources_domain", "domain"),
Index("ix_sources_parent_source_id", "parent_source_id"),
)
# ---------------------------------------------------------------------------
# Claim
# ---------------------------------------------------------------------------
class ClaimModel(Base):
__tablename__ = "claims"
id = Column(String(36), primary_key=True, default=lambda: str(uuid4()))
source_id = Column(String(36), ForeignKey("sources.id"), nullable=False)
claim = Column(Text, nullable=False)
normalized_claim = Column(Text, nullable=True)
claim_type = Column(Enum(ClaimType), nullable=False, default=ClaimType.FACTUAL)
subject = Column(Text, nullable=True)
predicate = Column(Text, nullable=True)
object = Column(Text, nullable=True)
evidence_span = Column(Text, nullable=True)
confidence = Column(Float, nullable=False, default=1.0)
event_date = Column(DateTime, nullable=True)
# Relationships
source = relationship("SourceModel", back_populates="claims")
# ---------------------------------------------------------------------------
# EvidenceRelation
# ---------------------------------------------------------------------------
class EvidenceRelationModel(Base):
__tablename__ = "evidence_relations"
id = Column(String(36), primary_key=True, default=lambda: str(uuid4()))
claim_a = Column(String(36), ForeignKey("claims.id"), nullable=False)
claim_b = Column(String(36), ForeignKey("claims.id"), nullable=False)
relation = Column(Enum(EvidenceRelationType), nullable=False)
confidence = Column(Float, nullable=False, default=0.5)
reason = Column(Text, nullable=True)
__table_args__ = (
Index("ix_evidence_relations_claim_a", "claim_a"),
Index("ix_evidence_relations_claim_b", "claim_b"),
# Ensure uniqueness of the (a, b, relation) triple
Index("uq_evidence_relations_ab", "claim_a", "claim_b", "relation", unique=True),
)
# ---------------------------------------------------------------------------
# CitationEdge
# ---------------------------------------------------------------------------
class CitationEdgeModel(Base):
__tablename__ = "citation_edges"
id = Column(String(36), primary_key=True, default=lambda: str(uuid4()))
source_from = Column(String(36), ForeignKey("sources.id"), nullable=False)
source_to = Column(String(36), ForeignKey("sources.id"), nullable=False)
relation = Column(Enum(EdgeRelation), nullable=False)
confidence = Column(Float, nullable=False, default=1.0)
__table_args__ = (
Index("ix_citation_edges_source_from", "source_from"),
Index("ix_citation_edges_source_to", "source_to"),
)
# ---------------------------------------------------------------------------
# ResearchReport
# ---------------------------------------------------------------------------
class ResearchReportModel(Base):
__tablename__ = "research_reports"
id = Column(String(36), primary_key=True, default=lambda: str(uuid4()))
research_id = Column(String(36), ForeignKey("search_queries.research_id"), nullable=False)
query = Column(Text, nullable=False)
summary = Column(Text, nullable=False, default="")
findings = Column(Text, nullable=False, default="[]")
disagreements = Column(Text, nullable=False, default="[]")
uncertainties = Column(Text, nullable=False, default="[]")
source_statistics = Column(Text, nullable=False, default="{}")
methodology = Column(Text, nullable=False, default="")
generated_at = Column(DateTime, nullable=False, default=datetime.utcnow)
__table_args__ = (
Index("ix_research_reports_research_id", "research_id"),
)

50
tests/conftest.py Normal file
View File

@@ -0,0 +1,50 @@
"""pytest fixtures for NSCT tests."""
from __future__ import annotations
import os
from typing import AsyncGenerator
import pytest
import pytest_asyncio
from fastapi.testclient import TestClient
from httpx import AsyncClient
from nsct.api.main import create_app
from nsct.config import AppSettings
@pytest.fixture(scope="session")
def app() -> AppSettings:
"""Minimal AppSettings for test environment."""
return AppSettings.from_env()
@pytest.fixture(scope="session")
def client(app: AppSettings):
"""Synchronous test client for the FastAPI app."""
fastapi_app = create_app()
with TestClient(fastapi_app) as c:
yield c
@pytest_asyncio.fixture(scope="function")
async def async_client(app: AppSettings) -> AsyncGenerator[AsyncClient, None]:
"""Async test client for the FastAPI app."""
fastapi_app = create_app()
async with AsyncClient(app=fastapi_app, base_url="http://test") as ac:
yield ac
@pytest.fixture(scope="function")
def clean_env(monkeypatch: pytest.MonkeyPatch) -> None:
"""Ensure environment has the minimum required vars for AppSettings."""
monkeypatch.setenv("NSCT_LLM_BASE_URL", "http://localhost:8030/openai/v1")
monkeypatch.setenv("NSCT_LLM_MODEL", "test-model")
monkeypatch.setenv("NSCT_LLM_MAX_CONCURRENCY", "1")
monkeypatch.setenv("NSCT_VISION_BASE_URL", "http://localhost:8030/openai/visual/v1")
monkeypatch.setenv("NSCT_VISION_MODEL", "test-model")
monkeypatch.setenv("NSCT_AUDIO_BASE_URL", "http://localhost:8030/hermes-audio")
monkeypatch.setenv("NSCT_AUDIO_MODEL", "default")
monkeypatch.setenv("NSCT_DB_URL", "sqlite+aiosqlite:///:memory:")
monkeypatch.setenv("NSCT_DEBUG", "false")

78
tests/test_health.py Normal file
View File

@@ -0,0 +1,78 @@
"""Tests for the health endpoints (/health, /ready, /providers)."""
from __future__ import annotations
from typing import Generator
import pytest
from fastapi.testclient import TestClient
@pytest.fixture
def client() -> Generator[TestClient, None, None]:
"""Synchronous test client."""
from nsct.api.main import create_app
app = create_app()
with TestClient(app) as c:
yield c
# ---------------------------------------------------------------------------
# /health
# ---------------------------------------------------------------------------
def test_health_returns_ok(client: TestClient) -> None:
"""GET /health should return status ok and the version."""
resp = client.get("/health")
assert resp.status_code == 200
body = resp.json()
assert body["status"] == "ok"
assert "version" in body
assert body["version"] == "0.1.0"
# ---------------------------------------------------------------------------
# /ready
# ---------------------------------------------------------------------------
def test_ready_returns_response(client: TestClient) -> None:
"""GET /ready should return a dict with status, llm key, etc."""
resp = client.get("/ready")
assert resp.status_code == 200
body = resp.json()
assert "status" in body
assert "llm" in body
def test_ready_status_field(client: TestClient) -> None:
"""GET /ready status should be 'ready' or 'not_ready'."""
resp = client.get("/ready")
body = resp.json()
assert body["status"] in ("ready", "not_ready")
# ---------------------------------------------------------------------------
# /providers
# ---------------------------------------------------------------------------
def test_providers_returns_provider_info(client: TestClient) -> None:
"""GET /providers should return keys for llm, vision, audio."""
resp = client.get("/providers")
assert resp.status_code == 200
body = resp.json()
assert "llm" in body
assert "vision" in body
assert "audio" in body
def test_providers_no_secrets(client: TestClient) -> None:
"""GET /providers must not contain secrets."""
resp = client.get("/providers")
body = resp.json()
flat = str(body)
assert "API_KEY" not in flat
assert "secret" not in flat.lower().replace("available", "")