diff --git a/README.de.md b/README.de.md
index 02f7c16..a550bf4 100644
--- a/README.de.md
+++ b/README.de.md
@@ -250,10 +250,3 @@ Siehe [CHANGELOG.md](CHANGELOG.md) fΓΌr Versionshistorie und Updates.
[π¬ Diskussionen](https://github.com/TimInTech/Pi-hole-Unbound-PiAlert-Setup/discussions)
-# Pi-hole + Unbound + NetAlertX β Ein-Klick-Setup
-
-
-
-- **Issues**: [GitHub Issues](https://github.com/TimInTech/Pi-hole-Unbound-PiAlert-Setup/issues)
-- **Diskussionen**: [GitHub Discussions](https://github.com/TimInTech/Pi-hole-Unbound-PiAlert-Setup/discussions)
-- **Dokumentation**: Diese README und Inline-Code-Kommentare
diff --git a/README.md b/README.md
index 2d3967a..d5a84b1 100644
--- a/README.md
+++ b/README.md
@@ -1,5 +1,251 @@
->>>>>> main
+
-- **Issues**: [GitHub Issues](https://github.com/TimInTech/Pi-hole-Unbound-PiAlert-Setup/issues)
-- **Discussions**: [GitHub Discussions](https://github.com/TimInTech/Pi-hole-Unbound-PiAlert-Setup/discussions)
-- **Documentation**: This README and inline code comments
\ No newline at end of file
+# π‘οΈ Pi-hole + Unbound + NetAlertX
+### **One-Click DNS Security & Monitoring Stack**
+
+[](https://github.com/TimInTech/Pi-hole-Unbound-PiAlert-Setup/actions)
+[](LICENSE)
+[](https://pi-hole.net/)
+[](https://nlnetlabs.nl/projects/unbound/)
+[](https://github.com/jokob-sk/NetAlertX)
+[](https://debian.org/)
+[](https://python.org/)
+
+**π§° Tech Stack**
+
+
+**π Languages:** π¬π§ English (this file) β’ [π©πͺ Deutsch](README.de.md)
+
+
+
+**Made with β€οΈ for the Pi-hole community**
+
+[π Report Bug](https://github.com/TimInTech/Pi-hole-Unbound-PiAlert-Setup/issues) β’
+[β¨ Request Feature](https://github.com/TimInTech/Pi-hole-Unbound-PiAlert-Setup/issues) β’
+[π¬ Discussions](https://github.com/TimInTech/Pi-hole-Unbound-PiAlert-Setup/discussions)
+
+
diff --git a/api/__init__.py b/api/__init__.py
index 57a924d..8ba89ed 100644
--- a/api/__init__.py
+++ b/api/__init__.py
@@ -1,2 +1 @@
-"""FastAPI API module."""
"""API package for the Pi-hole suite."""
diff --git a/api/main.py b/api/main.py
index 675f74d..9da9ba5 100644
--- a/api/main.py
+++ b/api/main.py
@@ -12,10 +12,10 @@
from shared.db import init_db
from .schemas import (
DeviceResponse,
- DNSLogResponse,
+ DNSLogResponse,
HealthResponse,
LeaseResponse,
- StatsResponse
+ StatsResponse,
)
@@ -26,17 +26,15 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
yield
-# FastAPI app with modern configuration
app = FastAPI(
title="Pi-hole Suite API",
description="Monitoring and management API for Pi-hole + Unbound + NetAlertX stack",
version="1.0.0",
docs_url="/docs",
redoc_url="/redoc",
- lifespan=lifespan
+ lifespan=lifespan,
)
-# Add CORS middleware
app.add_middleware(
CORSMiddleware,
allow_origins=["http://localhost:3000", "http://127.0.0.1:3000"],
@@ -47,19 +45,12 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
def get_api_key() -> str:
- """Get configured API key."""
-"""FastAPI app exposing Pi-hole suite data."""
-import os
-import sqlite3
-
-
-
-def _get_api_key() -> str:
+ """Get configured API key from environment."""
return os.getenv("SUITE_API_KEY", "")
def get_db() -> Generator[sqlite3.Connection, None, None]:
- """Database dependency."""
+ """FastAPI dependency to provide a DB connection."""
conn = sqlite3.connect(config.DB_PATH)
conn.row_factory = sqlite3.Row
try:
@@ -72,72 +63,56 @@ def require_api_key(x_api_key: Optional[str] = Header(None)) -> None:
"""Require valid API key for authentication."""
api_key = get_api_key()
if not api_key:
- raise HTTPException(
- status_code=500,
- detail="API key not configured on server"
- )
+ raise HTTPException(status_code=500, detail="API key not configured on server")
if not x_api_key:
- raise HTTPException(
- status_code=401,
- detail="Missing API key header"
- )
+ raise HTTPException(status_code=401, detail="Missing API key header")
if x_api_key != api_key:
- raise HTTPException(
- status_code=401,
- detail="Invalid API key"
- )
+ raise HTTPException(status_code=401, detail="Invalid API key")
@app.get("/", include_in_schema=False)
async def root():
- """Root endpoint redirect."""
+ """Root endpoint information."""
return {"message": "Pi-hole Suite API", "docs": "/docs"}
@app.get(
- "/health",
- dependencies=[Depends(require_api_key)],
+ "/health",
+ dependencies=[Depends(require_api_key)],
response_model=HealthResponse,
summary="Health Check",
- description="Simple health check endpoint to verify API is running"
+ description="Simple health check endpoint to verify API is running",
)
def health() -> HealthResponse:
- """Health check endpoint."""
- return HealthResponse(
- ok=True,
- message="Pi-hole Suite API is running",
- version="1.0.0"
- )
+ return HealthResponse(ok=True, message="Pi-hole Suite API is running", version="1.0.0")
@app.get(
- "/dns",
- dependencies=[Depends(require_api_key)],
+ "/dns",
+ dependencies=[Depends(require_api_key)],
response_model=List[DNSLogResponse],
summary="DNS Query Logs",
- description="Get recent DNS query logs from Pi-hole"
+ description="Get recent DNS query logs from Pi-hole",
)
def get_dns_logs(
limit: int = Query(50, ge=1, le=1000, description="Maximum number of logs to return"),
- db: sqlite3.Connection = Depends(get_db)
+ db: sqlite3.Connection = Depends(get_db),
) -> List[DNSLogResponse]:
- """Get recent DNS query logs."""
cursor = db.execute(
"SELECT timestamp, client, query, action FROM dns_logs ORDER BY id DESC LIMIT ?",
- (limit,)
+ (limit,),
)
return [DNSLogResponse(**dict(row)) for row in cursor.fetchall()]
@app.get(
- "/leases",
+ "/leases",
dependencies=[Depends(require_api_key)],
response_model=List[LeaseResponse],
- summary="DHCP Leases",
- description="Get current DHCP lease information"
+ summary="DHCP Leases",
+ description="Get current DHCP lease information",
)
def get_ip_leases(db: sqlite3.Connection = Depends(get_db)) -> List[LeaseResponse]:
- """Get IP lease information."""
cursor = db.execute(
"SELECT ip, mac, hostname, lease_start, lease_end FROM ip_leases ORDER BY lease_start DESC"
)
@@ -145,14 +120,13 @@ def get_ip_leases(db: sqlite3.Connection = Depends(get_db)) -> List[LeaseRespons
@app.get(
- "/devices",
- dependencies=[Depends(require_api_key)],
+ "/devices",
+ dependencies=[Depends(require_api_key)],
response_model=List[DeviceResponse],
summary="Network Devices",
- description="Get list of known network devices"
+ description="Get list of known network devices",
)
def get_devices(db: sqlite3.Connection = Depends(get_db)) -> List[DeviceResponse]:
- """Get network devices list."""
cursor = db.execute(
"SELECT id, ip, mac, hostname, last_seen FROM devices ORDER BY last_seen DESC"
)
@@ -164,42 +138,14 @@ def get_devices(db: sqlite3.Connection = Depends(get_db)) -> List[DeviceResponse
dependencies=[Depends(require_api_key)],
response_model=StatsResponse,
summary="System Statistics",
- description="Get basic statistics about the monitored system"
+ description="Get basic statistics about the monitored system",
)
def get_stats(db: sqlite3.Connection = Depends(get_db)) -> StatsResponse:
- """Get system statistics."""
- # Get DNS log count
- dns_cursor = db.execute("SELECT COUNT(*) as count FROM dns_logs")
- dns_count = dns_cursor.fetchone()["count"]
-
- # Get device count
- device_cursor = db.execute("SELECT COUNT(*) as count FROM devices")
- device_count = device_cursor.fetchone()["count"]
-
- # Get recent queries (last hour)
- recent_cursor = db.execute(
+ dns_count = db.execute("SELECT COUNT(*) as count FROM dns_logs").fetchone()["count"]
+ device_count = db.execute("SELECT COUNT(*) as count FROM devices").fetchone()["count"]
+ recent_queries = db.execute(
"SELECT COUNT(*) as count FROM dns_logs WHERE timestamp > datetime('now', '-1 hour')"
- )
- recent_queries = recent_cursor.fetchone()["count"]
-
- return StatsResponse(
- total_dns_logs=dns_count,
- total_devices=device_count,
- recent_queries=recent_queries
- )
-
-
- cur = db.execute(
- "SELECT timestamp, client, query, action FROM dns_logs ORDER BY id DESC LIMIT ?",
- (limit,),
- )
-
-
-
-@app.get("/leases", dependencies=[Depends(require_key)])
-def get_ip_leases(db=Depends(get_db)):
- cur = db.execute("SELECT ip, mac, hostname, lease_start, lease_end FROM ip_leases")
- return [dict(row) for row in cur.fetchall()]
-
+ ).fetchone()["count"]
+ return StatsResponse(total_dns_logs=dns_count, total_devices=device_count, recent_queries=recent_queries)
diff --git a/api/schemas.py b/api/schemas.py
index 08e10f7..cea1886 100644
--- a/api/schemas.py
+++ b/api/schemas.py
@@ -1,19 +1,18 @@
-"""Pydantic schemas for API responses."""
+"""Pydantic models for API request/response validation."""
+import ipaddress
from typing import Optional
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, field_validator
class HealthResponse(BaseModel):
- """Health check response."""
ok: bool = Field(..., description="Whether the service is healthy")
message: Optional[str] = Field(None, description="Optional status message")
version: Optional[str] = Field(None, description="API version")
class DNSLogResponse(BaseModel):
- """DNS log entry response."""
timestamp: str = Field(..., description="Query timestamp")
client: str = Field(..., description="Client IP address")
query: str = Field(..., description="DNS query domain")
@@ -21,7 +20,6 @@ class DNSLogResponse(BaseModel):
class LeaseResponse(BaseModel):
- """DHCP lease response."""
ip: str = Field(..., description="IP address")
mac: str = Field(..., description="MAC address")
hostname: Optional[str] = Field(None, description="Device hostname")
@@ -30,7 +28,6 @@ class LeaseResponse(BaseModel):
class DeviceResponse(BaseModel):
- """Network device response."""
id: int = Field(..., description="Device ID")
ip: str = Field(..., description="IP address")
mac: str = Field(..., description="MAC address")
@@ -39,62 +36,33 @@ class DeviceResponse(BaseModel):
class StatsResponse(BaseModel):
- """System statistics response."""
total_dns_logs: int = Field(..., description="Total DNS log entries")
total_devices: int = Field(..., description="Total known devices")
recent_queries: int = Field(..., description="Queries in the last hour")
-"""Pydantic models for API request/response validation."""
-from typing import Optional
-from pydantic import BaseModel, Field, field_validator
-import ipaddress
class DeviceRequest(BaseModel):
- """Request model for device operations."""
+ """Request model for device operations with validation."""
ip_address: str = Field(..., description="IP address of the device")
status: bool = Field(..., description="Device status (active/inactive)")
hostname: Optional[str] = Field(None, description="Device hostname")
mac_address: Optional[str] = Field(None, description="MAC address")
-
- @field_validator('ip_address')
+
+ @field_validator("ip_address")
@classmethod
- def validate_ip_address(cls, v):
- """Validate IP address format."""
+ def validate_ip_address(cls, v: str) -> str:
try:
ipaddress.ip_address(v)
- except ValueError:
- raise ValueError('Invalid IP address format')
+ except ValueError as exc: # noqa: TRY003 - re-raise as ValueError for pydantic
+ raise ValueError("Invalid IP address format") from exc
return v
-
- @field_validator('mac_address')
+
+ @field_validator("mac_address")
@classmethod
- def validate_mac_address(cls, v):
- """Validate MAC address format if provided."""
+ def validate_mac_address(cls, v: Optional[str]) -> Optional[str]:
if v is None:
return v
- # Simple MAC address validation (XX:XX:XX:XX:XX:XX format)
- if not (len(v) == 17 and v.count(':') == 5):
- raise ValueError('Invalid MAC address format')
+ # Simple MAC address validation (XX:XX:XX:XX:XX:XX)
+ if not (len(v) == 17 and v.count(":") == 5):
+ raise ValueError("Invalid MAC address format")
return v.lower()
-
-
-class DeviceResponse(BaseModel):
- """Response model for device data."""
- id: Optional[int] = None
- ip: str
- mac: Optional[str] = None
- hostname: Optional[str] = None
- last_seen: Optional[str] = None
-
-
-class DNSLogResponse(BaseModel):
- """Response model for DNS log entries."""
- timestamp: str
- client: str
- query: str
- action: str
-
-
-class HealthResponse(BaseModel):
- """Response model for health check."""
- ok: bool
diff --git a/install.sh b/install.sh
index 420ea24..7efdb6e 100755
--- a/install.sh
+++ b/install.sh
@@ -2,10 +2,7 @@
set -euo pipefail
# Pi-hole + Unbound + NetAlertX + Python Suite One-Click Installer
-# Modern, idempotent installer for complete DNS security stack
-#
-# Author: TimInTech
-# License: MIT
+# Author: TimInTech | License: MIT
# Repository: https://github.com/TimInTech/Pi-hole-Unbound-PiAlert-Setup
# π§ Configuration
@@ -13,305 +10,77 @@ readonly UNBOUND_PORT=5335
readonly NETALERTX_PORT=20211
readonly PYTHON_SUITE_PORT=8090
readonly NETALERTX_IMAGE="techxartisan/netalertx:latest"
+readonly SUITE_API_KEY=${SUITE_API_KEY:-$(openssl rand -hex 16)}
+readonly INSTALL_USER=${SUDO_USER:-$(whoami)}
+readonly INSTALL_HOME=$(getent passwd "$INSTALL_USER" | cut -d: -f6)
+readonly PROJECT_DIR="$(pwd)"
# π¨ Colors
-readonly RED='\033[0;31m'
-readonly GREEN='\033[0;32m'
-readonly YELLOW='\033[1;33m'
-readonly BLUE='\033[0;34m'
-readonly PURPLE='\033[0;35m'
-readonly CYAN='\033[0;36m'
-readonly NC='\033[0m'
+RED='\033[0;31m'; GREEN='\033[0;32m'; YELLOW='\033[1;33m'; BLUE='\033[0;34m'; NC='\033[0m'
-# π Logging functions
-log() { echo -e "${BLUE}[INFO]${NC} $*"; }
-success() { echo -e "${GREEN}[β]${NC} $*"; }
-warn() { echo -e "${YELLOW}[β ]${NC} $*"; }
-error() { echo -e "${RED}[β]${NC} $*"; }
-step() { echo -e "\n${PURPLE}[STEP]${NC} $*"; }
+# π Logging helpers
+log() { echo -e "${BLUE}[INFO]${NC} $*"; }
+success() { echo -e "${GREEN}[β]${NC} $*"; }
+warn() { echo -e "${YELLOW}[β ]${NC} $*"; }
+error() { echo -e "${RED}[β]${NC} $*"; }
+step() { echo -e "\n${YELLOW}[STEP]${NC} $*"; }
# π‘οΈ Error handler
-cleanup() {
- if [[ $? -ne 0 ]]; then
- error "Installation failed! Check logs above."
- error "You can re-run this script to retry installation."
- fi
-}
-trap cleanup EXIT
+trap 'error "Installation failed. See logs above."' ERR
+# ---------------------------------------------------------------------------
# π System checks
check_system() {
step "Performing system checks"
-
- # Check if running as root
- if [[ $EUID -ne 0 ]]; then
- error "This script must be run as root or with sudo"
- exit 1
- fi
-
- # Check for Debian/Ubuntu
- if ! command -v apt-get >/dev/null 2>&1; then
- error "This installer requires Debian/Ubuntu (apt-get not found)"
- exit 1
- fi
-
- # Detect OS
- if [[ -f /etc/os-release ]]; then
- . /etc/os-release
- log "Detected: $PRETTY_NAME"
- fi
-
- # Check project directory
- if [[ ! -f requirements.txt ]]; then
- error "requirements.txt not found. Please run from project directory"
- exit 1
- fi
-
- # Check internet connectivity
- if ! curl -s --connect-timeout 5 google.com >/dev/null; then
- warn "Internet connectivity check failed - continuing anyway"
- else
- success "Internet connectivity confirmed"
- fi
-
+ [[ $EUID -eq 0 ]] || { error "Run as root or with sudo"; exit 1; }
+ command -v apt-get >/dev/null || { error "Debian/Ubuntu required"; exit 1; }
+ [[ -f requirements.txt ]] || { error "Run script in project directory"; exit 1; }
success "System checks passed"
}
-# π Port conflict check
+# π Port conflicts
check_ports() {
- step "Checking port availability"
-
+ step "Checking ports"
local ports=($UNBOUND_PORT $NETALERTX_PORT $PYTHON_SUITE_PORT 53)
- local conflicts=()
-
for port in "${ports[@]}"; do
- if ss -tuln 2>/dev/null | grep -q ":$port "; then
- conflicts+=($port)
+ if ss -tuln | grep -q ":$port "; then
+ warn "Port $port already in use"
fi
done
-
- if [[ ${#conflicts[@]} -gt 0 ]]; then
- warn "Ports in use: ${conflicts[*]}"
- warn "Installation will continue but may require manual intervention"
- else
- success "All required ports available"
- fi
}
-# π¦ Package installation
+# π¦ Packages
install_packages() {
step "Installing system packages"
-
- log "Updating package lists..."
apt-get update -qq
-
- log "Installing core packages..."
- apt-get install -y \
- unbound \
- unbound-anchor \
-# This script installs and configures:
-# - Unbound DNS resolver on 127.0.0.1:5335
-# - Pi-hole using Unbound as upstream
-# - NetAlertX as Docker container on port 20211
-# - Python monitoring suite as systemd service
-
-# Configuration
-UNBOUND_PORT=5335
-NETALERTX_PORT=20211
-PYTHON_SUITE_PORT=8090
-SUITE_API_KEY=${SUITE_API_KEY:-$(openssl rand -hex 16)}
-INSTALL_USER=${SUDO_USER:-$(whoami)}
-INSTALL_HOME=$(getent passwd "$INSTALL_USER" | cut -d: -f6)
-PROJECT_DIR="$INSTALL_HOME/Pi-hole-Unbound-PiAlert-Setup"
-
-# Colors for output
-RED='\033[0;31m'
-GREEN='\033[0;32m'
-YELLOW='\033[1;33m'
-BLUE='\033[0;34m'
-NC='\033[0m' # No Color
-
-# Logging functions
-log_info() {
- echo -e "${BLUE}[INFO]${NC} $1"
-}
-
-log_success() {
- echo -e "${GREEN}[SUCCESS]${NC} $1"
-}
-
-log_warning() {
- echo -e "${YELLOW}[WARNING]${NC} $1"
-}
-
-log_error() {
- echo -e "${RED}[ERROR]${NC} $1"
-}
-
-# Check if running as root or with sudo
-check_privileges() {
- if [[ $EUID -ne 0 ]]; then
- log_error "This script must be run as root or with sudo"
- exit 1
- fi
-}
-
-# System checks
-check_system() {
- log_info "Performing system checks..."
-
- # Check if this is Debian/Ubuntu
- if ! command -v apt-get &> /dev/null; then
- log_error "This script requires a Debian/Ubuntu system with apt-get"
- exit 1
- fi
-
- # Check system version
- if [[ -f /etc/os-release ]]; then
- . /etc/os-release
- log_info "Detected system: $NAME $VERSION"
- fi
-
- # Check if we're in the project directory
- if [[ ! -f "start_suite.py" || ! -f "requirements.txt" ]]; then
- log_error "This script must be run from the Pi-hole-Unbound-PiAlert-Setup directory"
- log_info "Please cd to the project directory and run: sudo ./install.sh"
- exit 1
- fi
-
- PROJECT_DIR=$(pwd)
- log_info "Project directory: $PROJECT_DIR"
-}
-
-# Check for port conflicts
-check_ports() {
- log_info "Checking for port conflicts..."
-
- local ports_to_check=("$UNBOUND_PORT" "$NETALERTX_PORT" "$PYTHON_SUITE_PORT")
- local conflicts=false
-
- for port in "${ports_to_check[@]}"; do
- if ss -tuln | grep -q ":$port "; then
- log_warning "Port $port is already in use"
- conflicts=true
- fi
- done
-
- if [[ "$conflicts" == true ]]; then
- log_warning "Some ports are in use. The installer will attempt to work around this."
- log_info "You may need to stop conflicting services manually if issues arise."
- fi
-}
-
-# Install system packages
-install_packages() {
- log_info "Installing system packages..."
-
- # Update package lists
- apt-get update
-
- # Install required packages
- apt-get install -y \
- unbound \
- ca-certificates \
- curl \
- dnsutils \
- python3 \
- python3-venv \
- python3-pip \
- git \
- docker.io \
- openssl \
- systemd \
- sqlite3
-
+ apt-get install -y unbound unbound-anchor ca-certificates curl dnsutils \
+ python3 python3-venv python3-pip git docker.io openssl systemd sqlite3
success "System packages installed"
}
-# π Unbound configuration
+# π Unbound config
configure_unbound() {
- step "Configuring Unbound DNS resolver"
-
- # Create unbound directory
+ step "Configuring Unbound"
install -d -m 0755 /var/lib/unbound
-
- # Download root hints
- log "Downloading DNS root hints..."
curl -fsSL https://www.internic.net/domain/named.root -o /var/lib/unbound/root.hints
-
- # Create Pi-hole configuration
- log "Creating Unbound configuration..."
- systemctl
-
- log_success "System packages installed"
-}
-# Configure Unbound
-configure_unbound() {
- log_info "Configuring Unbound DNS resolver..."
-
- # Create unbound directory if it doesn't exist
- install -d -m 0755 /var/lib/unbound
-
- # Download root hints
- if [[ ! -f /var/lib/unbound/root.hints ]] || [[ $(find /var/lib/unbound/root.hints -mtime +30 2>/dev/null) ]]; then
- log_info "Downloading DNS root hints..."
- curl -fsSL https://www.internic.net/domain/named.root -o /var/lib/unbound/root.hints
- fi
-
- # Create Pi-hole specific Unbound configuration
- cat > /etc/unbound/unbound.conf.d/pi-hole.conf << 'EOF'
+ cat > /etc/unbound/unbound.conf.d/pi-hole.conf </dev/null \
+ && success "Unbound OK" || warn "Unbound may not be working"
}
-# π³οΈ Pi-hole installation
+# π³οΈ Pi-hole
install_pihole() {
step "Installing Pi-hole"
-
- if command -v pihole >/dev/null 2>&1; then
- log "Pi-hole already installed"
- else
- log "Downloading Pi-hole installer..."
- # Create minimal setupVars for unattended install
- mkdir -p /etc/pihole
- cat > /etc/pihole/setupVars.conf << EOF
-PIHOLE_INTERFACE=eth0
-IPV4_ADDRESS=127.0.0.1/8
-IPV6_ADDRESS=
-PIHOLE_DNS_1=127.0.0.1#$UNBOUND_PORT
-PIHOLE_DNS_2=
-QUERY_LOGGING=true
-INSTALL_WEB_SERVER=true
-INSTALL_WEB_INTERFACE=true
-LIGHTTPD_ENABLED=true
-CACHE_SIZE=10000
-DNS_FQDN_REQUIRED=true
-DNS_BOGUS_PRIV=true
-DNSMASQ_LISTENING=local
-WEBPASSWORD=
-BLOCKING_ENABLED=true
-EOF
-
- # Run Pi-hole installer
+ if ! command -v pihole >/dev/null; then
curl -sSL https://install.pi-hole.net | bash /dev/stdin --unattended
fi
-
- # Ensure Pi-hole uses Unbound
- log "Configuring Pi-hole upstream DNS..."
- log_success "Unbound is working correctly"
- else
- log_warning "Unbound test failed, but continuing..."
- fi
-}
-# Install Pi-hole
-install_pihole() {
- log_info "Installing Pi-hole..."
-
- # Check if Pi-hole is already installed
- if command -v pihole &> /dev/null; then
- log_info "Pi-hole is already installed, configuring..."
- else
- log_info "Downloading and installing Pi-hole..."
- # Use Pi-hole's official installer
- curl -sSL https://install.pi-hole.net | bash /dev/stdin --unattended
- fi
-
- # Configure Pi-hole to use Unbound
- log_info "Configuring Pi-hole to use Unbound..."
-
- # Set upstream DNS to Unbound
- if [[ -f /etc/pihole/setupVars.conf ]]; then
- sed -i "s/^PIHOLE_DNS_1=.*/PIHOLE_DNS_1=127.0.0.1#$UNBOUND_PORT/" /etc/pihole/setupVars.conf
- sed -i "s/^PIHOLE_DNS_2=.*/PIHOLE_DNS_2=/" /etc/pihole/setupVars.conf
- fi
-
- # Restart Pi-hole
+ sed -i "s/^PIHOLE_DNS_1=.*/PIHOLE_DNS_1=127.0.0.1#$UNBOUND_PORT/" /etc/pihole/setupVars.conf
+ sed -i "s/^PIHOLE_DNS_2=.*/PIHOLE_DNS_2=/" /etc/pihole/setupVars.conf
pihole restartdns
-
- success "Pi-hole configured with Unbound upstream"
+ success "Pi-hole configured with Unbound"
}
-# π³ NetAlertX installation
+# π³ NetAlertX
install_netalertx() {
step "Installing NetAlertX"
-
- # Start Docker
systemctl enable docker
systemctl start docker
-
- # Create data directories
mkdir -p /opt/netalertx/{config,db}
-
- # Stop existing container
- docker stop netalertx 2>/dev/null || true
- docker rm netalertx 2>/dev/null || true
-
- # Run NetAlertX
- log "Starting NetAlertX container..."
- # Restart Pi-hole DNS
- pihole restartdns
-
- log_success "Pi-hole configured to use Unbound"
-}
-
-# Install NetAlertX
-install_netalertx() {
- log_info "Installing NetAlertX..."
-
- # Start Docker service
- systemctl enable docker
- systemctl start docker
-
- # Create NetAlertX data directory
- mkdir -p /opt/netalertx/{config,db}
- chown -R "$INSTALL_USER:$INSTALL_USER" /opt/netalertx
-
- # Stop any existing NetAlertX container
- docker stop netalertx 2>/dev/null || true
- docker rm netalertx 2>/dev/null || true
-
- # Run NetAlertX container
- docker run -d \
- --name netalertx \
- --restart unless-stopped \
+ docker rm -f netalertx 2>/dev/null || true
+ docker run -d --name netalertx --restart unless-stopped \
-p $NETALERTX_PORT:20211 \
-v /opt/netalertx/config:/app/config \
-v /opt/netalertx/db:/app/db \
- -e TZ=$(timedatectl show --property=Timezone --value 2>/dev/null || echo "UTC") \
$NETALERTX_IMAGE
-
- success "NetAlertX installed on port $NETALERTX_PORT"
+ success "NetAlertX running on :$NETALERTX_PORT"
}
-# π Python suite setup
+# π Python Suite
setup_python_suite() {
- step "Setting up Python monitoring suite"
-
- local project_dir=$(pwd)
- local install_user=${SUDO_USER:-$(logname 2>/dev/null || echo "root")}
-
- # Create virtual environment
- if [[ ! -d .venv ]]; then
- log "Creating Python virtual environment..."
- python3 -m venv .venv
- chown -R $install_user:$install_user .venv 2>/dev/null || true
- fi
-
- # Install dependencies
- log "Installing Python dependencies..."
- .venv/bin/pip install -U pip
- .venv/bin/pip install -r requirements.txt
-
- # Generate API key
- local api_key=$(openssl rand -hex 16)
-
- # Create environment file
- cat > .env << EOF
-# Pi-hole Suite Configuration
-SUITE_API_KEY=$api_key
-SUITE_DATA_DIR=$project_dir/data
-SUITE_LOG_LEVEL=INFO
-EOF
-
- # Create data directory
- mkdir -p data
- chown -R $install_user:$install_user . 2>/dev/null || true
-
- # Create systemd service
- log "Creating systemd service..."
- cat > /etc/systemd/system/pihole-suite.service << EOF
-[Unit]
-Description=Pi-hole Suite (API + monitoring)
- -e TZ=$(timedatectl show --property=Timezone --value) \
- jokobsk/netalertx:latest
-
- log_success "NetAlertX installed and running on port $NETALERTX_PORT"
-}
-
-# Setup Python suite
-setup_python_suite() {
- log_info "Setting up Python monitoring suite..."
-
- # Ensure we're in the project directory
+ step "Setting up Python suite"
cd "$PROJECT_DIR"
-
- # Create virtual environment
- if [[ ! -d .venv ]]; then
- sudo -u "$INSTALL_USER" python3 -m venv .venv
- fi
-
- # Install Python dependencies
+ [[ -d .venv ]] || sudo -u "$INSTALL_USER" python3 -m venv .venv
sudo -u "$INSTALL_USER" .venv/bin/pip install -U pip
sudo -u "$INSTALL_USER" .venv/bin/pip install -r requirements.txt
-
- # Initialize database
- sudo -u "$INSTALL_USER" .venv/bin/python scripts/bootstrap.py 2>/dev/null || true
-
- # Create systemd service
- cat > /etc/systemd/system/pihole-suite.service << EOF
+ sudo -u "$INSTALL_USER" .venv/bin/python scripts/bootstrap.py || true
+
+ cat > /etc/systemd/system/pihole-suite.service < .env </dev/null | cut -d= -f2 || echo 'Check .env file')"
- echo -e " β’ Config file: $(pwd)/.env"
- echo
- echo -e "π οΈ ${BLUE}Service management:${NC}"
- echo -e " β’ systemctl status pihole-suite"
- echo -e " β’ journalctl -u pihole-suite -f"
- echo -e " β’ docker logs netalertx"
- echo
- echo -e "π ${BLUE}Next steps:${NC}"
- echo -e " 1. Configure devices to use $(hostname -I | awk '{print $1}') as DNS"
- echo -e " 2. Access Pi-hole admin to review settings"
- echo -e " 3. Check NetAlertX for network monitoring"
- echo -e " 4. Test API: curl -H \"X-API-Key: \$SUITE_API_KEY\" http://127.0.0.1:$PYTHON_SUITE_PORT/health"
- echo
-}
-
-# π Main installation
-main() {
- echo -e "${CYAN}"
- echo "βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ"
- echo "β Pi-hole + Unbound + NetAlertX β"
- echo "β One-Click Installer β"
- echo "β β"
- echo "β π‘οΈ DNS Security β’ π Network Monitoring β’ π§ Python Suite β"
- echo "βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ"
- echo -e "${NC}"
-
- check_system
- check_ports
- log_success "Python suite installed and running"
+ dig +short @127.0.0.1 -p $UNBOUND_PORT example.com | grep -q "." && success "Unbound OK" || error "Unbound FAIL"
+ pihole status | grep -q "blocking is enabled" && success "Pi-hole OK" || warn "Pi-hole status unclear"
+ docker ps | grep -q netalertx && success "NetAlertX OK" || warn "NetAlertX missing"
+ systemctl is-active --quiet pihole-suite && success "Python suite OK" || warn "Python suite not active"
}
-# Health checks
-run_health_checks() {
- log_info "Running health checks..."
-
- local all_healthy=true
-
- # Test Unbound
- log_info "Testing Unbound..."
- if dig +short @127.0.0.1 -p $UNBOUND_PORT example.com | grep -q "."; then
- log_success "β Unbound is responding"
- else
- log_error "β Unbound is not responding"
- all_healthy=false
- fi
-
- # Test Pi-hole
- log_info "Testing Pi-hole..."
- if pihole status | grep -q "Pi-hole blocking is enabled"; then
- log_success "β Pi-hole is running"
- else
- log_warning "β Pi-hole status unclear"
- fi
-
- # Test NetAlertX
- log_info "Testing NetAlertX..."
- if curl -s -m 5 "http://127.0.0.1:$NETALERTX_PORT" >/dev/null 2>&1; then
- log_success "β NetAlertX is responding"
- else
- log_warning "β NetAlertX not responding (may still be starting)"
- fi
-
- # Test Python suite
- log_info "Testing Python suite..."
- sleep 5 # Give the service time to start
- if curl -s -H "X-API-Key: $SUITE_API_KEY" "http://127.0.0.1:$PYTHON_SUITE_PORT/health" | grep -q '"ok":true'; then
- log_success "β Python suite API is responding"
- else
- log_warning "β Python suite not responding (may still be starting)"
- fi
-
- if [[ "$all_healthy" == true ]]; then
- log_success "All components are healthy!"
- else
- log_warning "Some components may need manual attention"
- fi
-}
-
-# Display summary
+# π Summary
show_summary() {
echo
- echo " Installation Complete!"
- echo
- echo "π§ Components installed:"
- echo " β’ Unbound DNS resolver: 127.0.0.1:$UNBOUND_PORT"
- echo " β’ Pi-hole web interface: http://$(hostname -I | awk '{print $1}')/admin"
- echo " β’ NetAlertX dashboard: http://$(hostname -I | awk '{print $1}'):$NETALERTX_PORT"
- echo " β’ Python suite API: http://127.0.0.1:$PYTHON_SUITE_PORT"
- echo
- echo "π API Configuration:"
- echo " β’ API Key: $SUITE_API_KEY"
- echo " β’ Test command: curl -H \"X-API-Key: $SUITE_API_KEY\" http://127.0.0.1:$PYTHON_SUITE_PORT/health"
- echo
- echo "π Important paths:"
- echo " β’ Project directory: $PROJECT_DIR"
- echo " β’ Data directory: $PROJECT_DIR/data"
- echo " β’ Unbound config: /etc/unbound/unbound.conf.d/pi-hole.conf"
- echo " β’ Pi-hole config: /etc/pihole/"
- echo " β’ NetAlertX data: /opt/netalertx/"
- echo
- echo "π οΈ Service management:"
- echo " β’ systemctl status pihole-suite"
- echo " β’ systemctl status unbound"
- echo " β’ docker logs netalertx"
- echo
- echo "π Next steps:"
- echo " 1. Access Pi-hole admin at http://$(hostname -I | awk '{print $1}')/admin"
- echo " 2. Verify DNS settings point to 127.0.0.1#$UNBOUND_PORT"
- echo " 3. Configure your devices to use this Pi-hole as DNS server"
- echo " 4. Check NetAlertX for network device monitoring"
- echo
- echo "β οΈ Save this API key: $SUITE_API_KEY"
- echo
+ success "Installation Complete!"
+ echo "Pi-hole admin: http://$(hostname -I | awk '{print $1}')/admin"
+ echo "NetAlertX: http://$(hostname -I | awk '{print $1}'):$NETALERTX_PORT"
+ echo "Python Suite: http://127.0.0.1:$PYTHON_SUITE_PORT"
+ echo "API Key: $SUITE_API_KEY"
}
-# Main installation function
+# π Main
main() {
- echo "Pi-hole + Unbound + NetAlertX + Python Suite Installer"
- echo
-
- check_privileges
check_system
check_ports
-
- log_info "Starting installation..."
-
install_packages
configure_unbound
install_pihole
@@ -802,20 +200,5 @@ main() {
setup_python_suite
run_health_checks
show_summary
-
- success "Installation completed successfully! π"
}
-
-# Execute main function
-
- log_info "Running health checks..."
- run_health_checks
-
- show_summary
-
- log_success "Installation completed successfully!"
- echo "You can now configure your devices to use $(hostname -I | awk '{print $1}') as their DNS server."
-}
-
-# Run main function
-main "$@"
\ No newline at end of file
+main "$@"
diff --git a/pyhole/dns_monitor.py b/pyhole/dns_monitor.py
index a98c2c9..868e08d 100644
--- a/pyhole/dns_monitor.py
+++ b/pyhole/dns_monitor.py
@@ -3,8 +3,6 @@
import logging
import re
import sqlite3
-"""Simple Pi-hole log tailer with log rotation support."""
-import logging
import threading
import time
from pathlib import Path
@@ -216,88 +214,3 @@ def stop() -> None:
def is_running() -> bool:
"""Check if the monitor is currently running."""
return _monitor_thread is not None and _monitor_thread.is_alive()
-logger = logging.getLogger(__name__)
-
-PIHOLE_LOG = Path("/var/log/pihole.log")
-_stop_event = threading.Event()
-
-
-def parse_line(line: str) -> Optional[Tuple[str, str, str, str]]:
- """Parse a Pi-hole log line into components."""
- parts = line.strip().split()
- if len(parts) < 5:
- return None
- timestamp = " ".join(parts[:2])
- action = parts[2]
- client = parts[3].rstrip(":")
- query = parts[4]
- return timestamp, client, query, action
-
-
-def monitor(conn, log_path: Optional[Path] = None) -> None:
- """Monitor Pi-hole log file with log rotation support."""
- log_path = log_path or PIHOLE_LOG
- logger.info("Starting DNS monitor on %s", log_path)
-
- last_pos = 0
- last_inode = None
-
- while not _stop_event.is_set():
- try:
- if log_path.exists():
- # Check if log file was rotated (inode changed)
- current_stat = log_path.stat()
- current_inode = current_stat.st_ino
-
- if last_inode is not None and current_inode != last_inode:
- logger.info("Log rotation detected, resetting position")
- last_pos = 0
-
- last_inode = current_inode
-
- # Check if file was truncated
- if current_stat.st_size < last_pos:
- logger.info("Log file truncated, resetting position")
- last_pos = 0
-
- with log_path.open() as handle:
- handle.seek(last_pos)
- lines_processed = 0
-
- for line in handle:
- parsed = parse_line(line)
- if parsed:
- try:
- conn.execute(
- "INSERT INTO dns_logs(timestamp, client, query, action) VALUES(?,?,?,?)",
- parsed,
- )
- lines_processed += 1
- except Exception as e:
- logger.warning("Failed to insert DNS log entry: %s", e)
-
- if lines_processed > 0:
- conn.commit()
- logger.debug("Processed %d DNS log entries", lines_processed)
-
- last_pos = handle.tell()
- else:
- logger.warning("Pi-hole log file not found: %s", log_path)
-
- except Exception as e:
- logger.error("Error monitoring DNS log: %s", e)
-
- time.sleep(5)
-
-
-def start(conn):
- """Start the DNS monitor in a background thread."""
- _stop_event.clear()
- thread = threading.Thread(target=monitor, args=(conn,), daemon=True)
- thread.start()
- return thread
-
-
-def stop() -> None:
- """Stop the DNS monitor."""
- _stop_event.set()
diff --git a/scripts/bootstrap.py b/scripts/bootstrap.py
index ba885db..91fea3a 100755
--- a/scripts/bootstrap.py
+++ b/scripts/bootstrap.py
@@ -58,27 +58,3 @@ def main():
if __name__ == "__main__":
sys.exit(main())
-"""Check suite dependencies."""
-import importlib.util
-import sys
-
-REQUIRED = [
- "fastapi",
- "uvicorn",
- "pydantic",
- "sqlite3",
- "ipaddress",
-]
-
-
-def main() -> None:
- missing = [mod for mod in REQUIRED if importlib.util.find_spec(mod) is None]
- if missing:
- print("Missing: " + ", ".join(missing))
- print("Run: pip install -r requirements.txt")
- sys.exit(1)
- print("Dependencies OK")
-
-
-if __name__ == "__main__":
- main()
diff --git a/scripts/healthcheck.py b/scripts/healthcheck.py
index 2cd7cdd..73f1a86 100755
--- a/scripts/healthcheck.py
+++ b/scripts/healthcheck.py
@@ -100,22 +100,3 @@ def main():
if __name__ == "__main__":
sys.exit(main())
-"""Verify database connectivity."""
-import sqlite3
-import sys
-from pathlib import Path
-
-
-
-def main() -> None:
- try:
- with sqlite3.connect(config.DB_PATH) as conn:
- conn.execute("SELECT 1")
- print("Database healthy")
- except Exception as exc: # noqa: BLE001 - we want to show any failure
- print(f"Failed: {exc}")
- sys.exit(1)
-
-
-if __name__ == "__main__":
- main()
diff --git a/shared/db.py b/shared/db.py
index 3d48fdc..296d5d4 100644
--- a/shared/db.py
+++ b/shared/db.py
@@ -60,39 +60,6 @@
);
CREATE INDEX IF NOT EXISTS idx_stats_timestamp ON system_stats(timestamp);
CREATE INDEX IF NOT EXISTS idx_stats_metric ON system_stats(metric_name);
-"""SQLite helpers for the Pi-hole suite."""
-import sqlite3
-from pathlib import Path
-
-from .shared_config import DB_PATH
-
-SCHEMA = """
-CREATE TABLE IF NOT EXISTS dns_logs(
- id INTEGER PRIMARY KEY AUTOINCREMENT,
- timestamp TEXT,
- client TEXT,
- query TEXT,
- action TEXT
-);
-CREATE INDEX IF NOT EXISTS idx_dns_timestamp ON dns_logs(timestamp);
-
-CREATE TABLE IF NOT EXISTS ip_leases(
- id INTEGER PRIMARY KEY AUTOINCREMENT,
- ip TEXT,
- mac TEXT,
- hostname TEXT,
- lease_start TEXT,
- lease_end TEXT
-);
-
-CREATE TABLE IF NOT EXISTS devices(
- id INTEGER PRIMARY KEY AUTOINCREMENT,
- ip TEXT,
- mac TEXT,
- hostname TEXT,
- last_seen TEXT
-);
-CREATE INDEX IF NOT EXISTS idx_devices_ip ON devices(ip);
"""
@@ -120,8 +87,3 @@ def get_connection() -> sqlite3.Connection:
conn = sqlite3.connect(str(DB_PATH))
conn.row_factory = sqlite3.Row
return conn
- Path(DB_PATH).parent.mkdir(parents=True, exist_ok=True)
- conn = sqlite3.connect(DB_PATH, check_same_thread=False)
- conn.executescript(SCHEMA)
- conn.commit()
- return conn
diff --git a/shared/shared_config.py b/shared/shared_config.py
index d1dfc97..83a9a1e 100644
--- a/shared/shared_config.py
+++ b/shared/shared_config.py
@@ -1,6 +1,6 @@
"""Shared configuration for the Pi-hole suite."""
-"""Shared configuration helpers for the suite."""
+"""
import logging
import os
from pathlib import Path
@@ -29,13 +29,3 @@
# Module logger
logger = logging.getLogger(__name__)
logger.info(f"Configuration loaded - Data directory: {DATA_DIR}, Log level: {LOG_LEVEL}")
-INTERFACE = os.getenv("SUITE_INTERFACE", "eth0")
-DNS_PORT = int(os.getenv("SUITE_DNS_PORT", "5335"))
-LOG_LEVEL = os.getenv("SUITE_LOG_LEVEL", "INFO")
-DATA_DIR = Path(os.getenv("SUITE_DATA_DIR", "data"))
-
-DATA_DIR.mkdir(parents=True, exist_ok=True)
-
-DB_PATH = DATA_DIR / "shared.sqlite"
-
-logging.basicConfig(level=LOG_LEVEL, format="%(asctime)s [%(levelname)s] %(message)s")
diff --git a/start_suite.py b/start_suite.py
index 65e5135..b6d95ad 100644
--- a/start_suite.py
+++ b/start_suite.py
@@ -28,7 +28,6 @@
# Optional demo component - disabled by default
ENABLE_PYALLOC_DEMO = os.getenv("ENABLE_PYALLOC_DEMO", "false").lower() == "true"
-
if ENABLE_PYALLOC_DEMO:
try:
from pyalloc.main import start as alloc_start
@@ -42,33 +41,8 @@ async def run_api() -> None:
"""Run the FastAPI application."""
port = int(os.getenv("SUITE_PORT", "8090"))
host = os.getenv("SUITE_HOST", "127.0.0.1")
-
- config = uvicorn.Config(
- api_app,
- host=host,
- port=port,
- log_level="info",
- access_log=True
- )
-import asyncio
-import os
-import threading
-
-import uvicorn
-
-from api.main import app as api_app
-from pyhole.dns_monitor import start as dns_start
-from shared.db import init_db
-# Optional demo component - disabled by default for one-click installer
-ENABLE_PYALLOC_DEMO = os.getenv("ENABLE_PYALLOC_DEMO", "false").lower() == "true"
-
-if ENABLE_PYALLOC_DEMO:
- from pyalloc.main import start as alloc_start
-
-
-async def run_api() -> None:
- config = uvicorn.Config(api_app, host="127.0.0.1", port=8090, log_level="info")
+ config = uvicorn.Config(api_app, host=host, port=port, log_level="info", access_log=True)
server = uvicorn.Server(config)
await server.serve()
@@ -76,14 +50,14 @@ async def run_api() -> None:
def main() -> None:
"""Main application entry point."""
logger.info("Starting Pi-hole Suite...")
-
+
# Verify API key
api_key = os.environ.get("SUITE_API_KEY")
if not api_key:
logger.error("SUITE_API_KEY environment variable must be set")
logger.info("Generate one with: openssl rand -hex 16")
sys.exit(1)
-
+
# Initialize database
try:
conn = init_db()
@@ -91,20 +65,20 @@ def main() -> None:
except Exception as e:
logger.error(f"Failed to initialize database: {e}")
sys.exit(1)
-
+
# Start core DNS monitoring
try:
dns_thread = threading.Thread(
- target=dns_start,
- args=(conn,),
+ target=dns_start,
+ args=(conn,),
daemon=True,
- name="DNSMonitor"
+ name="DNSMonitor",
)
dns_thread.start()
logger.info("DNS monitor started")
except Exception as e:
logger.error(f"Failed to start DNS monitor: {e}")
-
+
# Start optional demo allocator if enabled
if ENABLE_PYALLOC_DEMO:
try:
@@ -112,28 +86,16 @@ def main() -> None:
target=alloc_start,
args=(conn,),
daemon=True,
- name="AllocDemo"
+ name="AllocDemo",
)
alloc_thread.start()
logger.info("PyAlloc demo component started")
except Exception as e:
logger.warning(f"Failed to start PyAlloc demo: {e}")
-
+
logger.info(f"API Key: {api_key[:8]}...")
logger.info("Starting API server...")
- conn = init_db()
-
- # Start core DNS monitoring
- threading.Thread(target=dns_start, args=(conn,), daemon=True).start()
-
- # Start optional demo allocator if enabled
- if ENABLE_PYALLOC_DEMO:
- threading.Thread(target=alloc_start, args=(conn,), daemon=True).start()
- print("β Started with pyalloc demo component")
- else:
- print("β Started in production mode (pyalloc demo disabled)")
-
try:
asyncio.run(run_api())
except KeyboardInterrupt:
@@ -143,10 +105,5 @@ def main() -> None:
sys.exit(1)
-if __name__ == "__main__":
- main()
- print("Shutting downβ¦")
-
-
if __name__ == "__main__":
main()
diff --git a/tests/test_api.py b/tests/test_api.py
index d210021..703fe57 100644
--- a/tests/test_api.py
+++ b/tests/test_api.py
@@ -7,207 +7,104 @@
from fastapi.testclient import TestClient
# Set test environment before importing
-os.environ['SUITE_API_KEY'] = 'test-api-key'
-os.environ['SUITE_DATA_DIR'] = tempfile.mkdtemp()
+os.environ["SUITE_API_KEY"] = "test-api-key"
+os.environ["SUITE_DATA_DIR"] = tempfile.mkdtemp()
from api.main import app
from shared.db import init_db
-@pytest.fixture(scope="module")
+@pytest.fixture(scope="module")
def setup_database():
- """Initialize test database once for all tests."""
init_db()
yield
- # Cleanup would happen here if needed
@pytest.fixture
def client(setup_database):
- """Create test client with initialized database."""
return TestClient(app)
@pytest.fixture
def api_headers():
- """Return API headers with test key."""
return {"X-API-Key": "test-api-key"}
+def test_root_endpoint(client):
+ resp = client.get("/")
+ assert resp.status_code == 200
+ assert "message" in resp.json()
+
+
def test_health_endpoint(client, api_headers):
- """Test the health endpoint."""
- response = client.get("/health", headers=api_headers)
- assert response.status_code == 200
- data = response.json()
+ resp = client.get("/health", headers=api_headers)
+ assert resp.status_code == 200
+ data = resp.json()
assert data["ok"] is True
assert "message" in data
-import pytest
-from fastapi.testclient import TestClient
-import os
-import tempfile
-import sqlite3
-from pathlib import Path
-
-from api.main import app
-
-
-@pytest.fixture
-def test_db():
- """Create a temporary test database."""
- with tempfile.NamedTemporaryFile(suffix='.db', delete=False) as tmp:
- test_db_path = tmp.name
-
- # Set environment variable for test database
- os.environ['SUITE_DATA_DIR'] = str(Path(test_db_path).parent)
- os.environ['SUITE_API_KEY'] = 'test-api-key'
-
- # Initialize test database
- conn = sqlite3.connect(test_db_path)
- conn.executescript("""
- CREATE TABLE IF NOT EXISTS devices(
- id INTEGER PRIMARY KEY AUTOINCREMENT,
- ip TEXT,
- mac TEXT,
- hostname TEXT,
- last_seen TEXT
- );
- CREATE TABLE IF NOT EXISTS dns_logs(
- id INTEGER PRIMARY KEY AUTOINCREMENT,
- timestamp TEXT,
- client TEXT,
- query TEXT,
- action TEXT
- );
- CREATE TABLE IF NOT EXISTS ip_leases(
- id INTEGER PRIMARY KEY AUTOINCREMENT,
- ip TEXT,
- mac TEXT,
- hostname TEXT,
- lease_start TEXT,
- lease_end TEXT
- );
- """)
- conn.commit()
- conn.close()
-
- yield test_db_path
-
- # Cleanup
- Path(test_db_path).unlink(missing_ok=True)
-
-
-@pytest.fixture
-def client(test_db):
- """Create test client with temporary database."""
- return TestClient(app)
-
-
-def test_health_endpoint(client):
- """Test the health endpoint."""
- response = client.get("/health", headers={"X-API-Key": "test-api-key"})
- assert response.status_code == 200
- assert response.json() == {"ok": True}
+ assert "version" in data
def test_health_endpoint_no_auth(client):
- """Test health endpoint without authentication."""
- response = client.get("/health")
- assert response.status_code == 401 # API key validation returns 401
+ resp = client.get("/health")
+ assert resp.status_code == 401
def test_health_endpoint_bad_auth(client):
- """Test health endpoint with bad authentication."""
- response = client.get("/health", headers={"X-API-Key": "wrong-key"})
- assert response.status_code == 401
+ resp = client.get("/health", headers={"X-API-Key": "wrong-key"})
+ assert resp.status_code == 401
def test_dns_logs_endpoint(client, api_headers):
- """Test DNS logs endpoint."""
- response = client.get("/dns", headers=api_headers)
- assert response.status_code == 422 # Unprocessable Entity due to missing required header
-
-
-def test_dns_logs_endpoint(client):
- """Test DNS logs endpoint."""
- response = client.get("/dns", headers={"X-API-Key": "test-api-key"})
- assert response.status_code == 200
- assert isinstance(response.json(), list)
+ resp = client.get("/dns", headers=api_headers)
+ assert resp.status_code == 200
+ assert isinstance(resp.json(), list)
def test_dns_logs_with_limit(client, api_headers):
- """Test DNS logs endpoint with limit parameter."""
- response = client.get("/dns?limit=10", headers=api_headers)
- assert response.status_code == 200
- data = response.json()
+ resp = client.get("/dns?limit=10", headers=api_headers)
+ assert resp.status_code == 200
+ data = resp.json()
assert isinstance(data, list)
assert len(data) <= 10
def test_devices_endpoint(client, api_headers):
- """Test devices endpoint."""
- response = client.get("/devices", headers=api_headers)
-def test_devices_endpoint(client):
- """Test devices endpoint."""
- response = client.get("/devices", headers={"X-API-Key": "test-api-key"})
- assert response.status_code == 200
- assert isinstance(response.json(), list)
+ resp = client.get("/devices", headers=api_headers)
+ assert resp.status_code == 200
+ assert isinstance(resp.json(), list)
def test_leases_endpoint(client, api_headers):
- """Test IP leases endpoint."""
- response = client.get("/leases", headers=api_headers)
-def test_leases_endpoint(client):
- """Test IP leases endpoint."""
- response = client.get("/leases", headers={"X-API-Key": "test-api-key"})
- assert response.status_code == 200
- assert isinstance(response.json(), list)
+ resp = client.get("/leases", headers=api_headers)
+ assert resp.status_code == 200
+ assert isinstance(resp.json(), list)
def test_stats_endpoint(client, api_headers):
- """Test statistics endpoint."""
- response = client.get("/stats", headers=api_headers)
- assert response.status_code == 200
- data = response.json()
+ resp = client.get("/stats", headers=api_headers)
+ assert resp.status_code == 200
+ data = resp.json()
assert "total_dns_logs" in data
assert "total_devices" in data
assert "recent_queries" in data
- assert isinstance(data["total_dns_logs"], int)
- assert isinstance(data["total_devices"], int)
- assert isinstance(data["recent_queries"], int)
-def test_root_endpoint(client):
- """Test root endpoint."""
- response = client.get("/")
- assert response.status_code == 200
- data = response.json()
- assert "message" in data
-def test_root_status():
- """Test basic application status."""
- # This is a placeholder test for basic functionality
- assert True # Replace with actual status checks
-
-
-# Example of how to test with the new Pydantic schemas
def test_device_request_validation():
- """Test DeviceRequest model validation."""
from api.schemas import DeviceRequest
-
- # Valid request
+
valid_data = {
"ip_address": "192.168.1.100",
"status": True,
"hostname": "test-device",
- "mac_address": "aa:bb:cc:dd:ee:ff"
+ "mac_address": "aa:bb:cc:dd:ee:ff",
}
device_req = DeviceRequest(**valid_data)
assert device_req.ip_address == "192.168.1.100"
assert device_req.status is True
-
- # Invalid IP address
+
with pytest.raises(ValueError):
DeviceRequest(ip_address="invalid-ip", status=True)
-
- # Invalid MAC address
+
with pytest.raises(ValueError):
DeviceRequest(ip_address="192.168.1.100", status=True, mac_address="invalid-mac")