Jarian Cottingham
7565366138
- Add URL-decode in validate_archive_path to block ..%2F encoded traversal - Rewrite test_storage_manager.py (was corrupted, #14) - Rewrite test_path_handling.py (remove hardcoded macOS path, #13) - Add test_web_interface.py (auth, CSRF, headers, XSS tests) - All 11 tests pass - Fixes: #1 path traversal, #2 XSS, #13 hardcoded path, #14 corrupted tests
701 lines
21 KiB
Python
701 lines
21 KiB
Python
#!/usr/bin/env python3
|
|
"""Web Interface for NewsArchiver - Phase 3
|
|
|
|
Flask web server for browsing archived news articles.
|
|
"""
|
|
|
|
import hashlib
|
|
import hmac
|
|
import html
|
|
import json
|
|
import logging
|
|
import os
|
|
import secrets
|
|
import sys
|
|
import xml.etree.ElementTree as ET
|
|
from datetime import datetime, timezone
|
|
from functools import wraps
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
from urllib.parse import quote, unquote, urlparse
|
|
|
|
from flask import (
|
|
Flask,
|
|
abort,
|
|
flash,
|
|
jsonify,
|
|
make_response,
|
|
redirect,
|
|
render_template,
|
|
request,
|
|
session,
|
|
url_for,
|
|
)
|
|
|
|
from storage_manager import (
|
|
DB_PATH,
|
|
get_all_sources,
|
|
get_article,
|
|
get_articles_by_source,
|
|
get_latest_articles,
|
|
get_source_stats,
|
|
)
|
|
|
|
SCRIPT_DIR = Path(__file__).parent.resolve()
|
|
ARCHIVE_DIR = Path(
|
|
os.environ.get("ARCHIVE_DIR", str(SCRIPT_DIR / "archival_data"))
|
|
).resolve()
|
|
RSS_FEEDS_PATH = SCRIPT_DIR / "rss_feeds.json"
|
|
|
|
RSS_FEEDS = {}
|
|
|
|
# Authentication configuration
|
|
ADMIN_PASSWORD = os.environ.get("NEWSARCHIVER_PASSWORD", "")
|
|
SECRET_KEY = os.environ.get(
|
|
"NEWSARCHIVER_SECRET_KEY", secrets.token_hex(32)
|
|
)
|
|
SESSION_LIFETIME_MINUTES = int(os.environ.get("NEWSARCHIVER_SESSION_MINUTES", "480"))
|
|
|
|
# Server URL configuration
|
|
SERVER_URL = os.environ.get(
|
|
"NEWSARCHIVER_SERVER_URL", f"http://localhost:5000"
|
|
)
|
|
|
|
app = Flask(
|
|
__name__,
|
|
static_folder=str(SCRIPT_DIR / "static"),
|
|
template_folder=str(SCRIPT_DIR / "templates"),
|
|
)
|
|
app.secret_key = SECRET_KEY
|
|
|
|
|
|
def login_required(f):
|
|
"""Decorator to require login for protected routes."""
|
|
|
|
@wraps(f)
|
|
def decorated_function(*args, **kwargs):
|
|
if not ADMIN_PASSWORD:
|
|
return f(*args, **kwargs)
|
|
if "authenticated" not in session:
|
|
flash("Please log in to access this page.", "error")
|
|
return redirect(url_for("login", next=request.url))
|
|
return f(*args, **kwargs)
|
|
|
|
return decorated_function
|
|
|
|
|
|
def generate_csrf_token():
|
|
"""Generate a CSRF token tied to the session."""
|
|
if "csrf_token" not in session:
|
|
session["csrf_token"] = secrets.token_hex(32)
|
|
return session["csrf_token"]
|
|
|
|
|
|
def verify_csrf_token():
|
|
"""Verify the CSRF token from the request."""
|
|
if not ADMIN_PASSWORD:
|
|
return True
|
|
token = request.form.get("csrf_token") or request.headers.get("X-CSRF-Token")
|
|
if not token or "csrf_token" not in session:
|
|
return False
|
|
return hmac.compare_digest(token, session["csrf_token"])
|
|
|
|
|
|
@app.template_global()
|
|
def csrf_token():
|
|
"""Make CSRF token available in templates."""
|
|
return generate_csrf_token()
|
|
|
|
|
|
@app.before_request
|
|
def enforce_session_timeout():
|
|
"""Expire sessions after inactivity."""
|
|
if "authenticated" in session:
|
|
auth_time = session.get("auth_time", 0)
|
|
now = datetime.now().timestamp()
|
|
if now - auth_time > SESSION_LIFETIME_MINUTES * 60:
|
|
session.clear()
|
|
|
|
|
|
@app.after_request
|
|
def add_security_headers(response):
|
|
"""Add security headers to all responses."""
|
|
response.headers["X-Content-Type-Options"] = "nosniff"
|
|
response.headers["X-Frame-Options"] = "DENY"
|
|
response.headers["X-XSS-Protection"] = "1; mode=block"
|
|
response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"
|
|
response.headers["Cache-Control"] = "no-store"
|
|
if not request.is_secure:
|
|
response.headers["Content-Security-Policy"] = (
|
|
"default-src 'self'; "
|
|
"script-src 'self'; "
|
|
"style-src 'self' 'unsafe-inline'; "
|
|
"img-src 'self' data:; "
|
|
"frame-ancestors 'none'"
|
|
)
|
|
return response
|
|
|
|
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format="%(asctime)s - %(levelname)s - %(message)s",
|
|
handlers=[
|
|
logging.StreamHandler(sys.stdout),
|
|
logging.FileHandler(ARCHIVE_DIR / "processing.log", encoding="utf-8"),
|
|
],
|
|
)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def get_pagination_info(total: int, page: int, per_page: int) -> dict:
|
|
"""Calculate pagination information.
|
|
|
|
Args:
|
|
total: Total number of items
|
|
page: Current page number
|
|
per_page: Items per page
|
|
|
|
Returns:
|
|
Dictionary with pagination details
|
|
"""
|
|
total_pages = (total + per_page - 1) // per_page if total > 0 else 1
|
|
|
|
return {
|
|
"total": total,
|
|
"page": page,
|
|
"per_page": per_page,
|
|
"has_next": page < total_pages,
|
|
"has_prev": page > 1,
|
|
"next_num": page + 1 if page < total_pages else None,
|
|
"prev_num": page - 1 if page > 1 else None,
|
|
"pages": total_pages,
|
|
}
|
|
|
|
|
|
def sanitize_error_message(msg: str) -> str:
|
|
"""Sanitize error messages to prevent information leakage."""
|
|
sanitized = msg
|
|
for pattern in [r"\d{1,3}(\.\d{1,3}){3}", r"/home/\w+"]:
|
|
import re
|
|
|
|
sanitized = re.sub(pattern, "[REDACTED]", sanitized)
|
|
return sanitized
|
|
|
|
|
|
@app.route("/login", methods=["GET", "POST"])
|
|
def login():
|
|
"""Login page."""
|
|
if not ADMIN_PASSWORD:
|
|
return redirect(url_for("index"))
|
|
|
|
if request.method == "POST":
|
|
if not verify_csrf_token():
|
|
flash("Invalid request.", "error")
|
|
return redirect(url_for("login"))
|
|
|
|
password = request.form.get("password", "")
|
|
if hmac.compare_digest(password, ADMIN_PASSWORD):
|
|
session.clear()
|
|
session["authenticated"] = True
|
|
session["csrf_token"] = secrets.token_hex(32)
|
|
session["auth_time"] = datetime.now().timestamp()
|
|
next_url = request.form.get("next", url_for("index"))
|
|
parsed = urlparse(next_url)
|
|
if parsed.netloc:
|
|
next_url = url_for("index")
|
|
return redirect(next_url)
|
|
else:
|
|
flash("Invalid password.", "error")
|
|
|
|
return render_template("login.html", csrf_token=csrf_token())
|
|
|
|
|
|
@app.route("/logout")
|
|
def logout():
|
|
"""Logout."""
|
|
session.clear()
|
|
return redirect(url_for("index"))
|
|
|
|
|
|
@app.route("/")
|
|
@login_required
|
|
def index():
|
|
"""Newspaper listing page."""
|
|
sources = get_all_sources()
|
|
rss_feeds = load_rss_feeds()
|
|
|
|
source_list = []
|
|
disabled_sources = []
|
|
|
|
for source_name in sources:
|
|
stats = get_source_stats(source_name)
|
|
|
|
source_info = {
|
|
"name": source_name.title(),
|
|
"slug": source_name,
|
|
"article_count": stats["total_articles"],
|
|
"last_archived": stats.get("last_archived"),
|
|
"status": "success" if stats["total_articles"] > 0 else "pending",
|
|
}
|
|
|
|
if source_name in rss_feeds:
|
|
feed_info = rss_feeds[source_name]
|
|
if feed_info.get("disabled", False):
|
|
source_info["disabled"] = True
|
|
source_info["disable_reason"] = feed_info.get(
|
|
"disable_reason", "No reason provided"
|
|
)
|
|
disabled_sources.append(source_info)
|
|
continue
|
|
|
|
source_list.append(source_info)
|
|
|
|
source_list.extend(disabled_sources)
|
|
|
|
return render_template("index.html", sources=source_list)
|
|
|
|
|
|
@app.route("/source/<slug>")
|
|
@login_required
|
|
def articles(slug: str):
|
|
"""Article listing page for a specific source."""
|
|
page = request.args.get("page", 1, type=int)
|
|
per_page = 50
|
|
|
|
sources = get_all_sources()
|
|
source_name = None
|
|
for s in sources:
|
|
if s.lower() == slug.lower():
|
|
source_name = s
|
|
break
|
|
|
|
if not source_name:
|
|
return render_template("article_not_found.html", slug=slug, article_id=0), 404
|
|
|
|
articles_list = get_articles_by_source(
|
|
source_name, limit=per_page, offset=(page - 1) * per_page
|
|
)
|
|
stats = get_source_stats(source_name)
|
|
total = stats["total_articles"]
|
|
|
|
pagination = get_pagination_info(total, page, per_page)
|
|
|
|
articles_data = []
|
|
for article in articles_list:
|
|
articles_data.append(
|
|
{
|
|
"id": getattr(article, "id", 0),
|
|
"title": article.title or "Untitled",
|
|
"date": article.publish_date or "",
|
|
"summary": article.content_text[:200] if article.content_text else "",
|
|
"url": f"/source/{source_name.lower()}/article/{getattr(article, 'id', 0)}",
|
|
}
|
|
)
|
|
|
|
return render_template(
|
|
"articles.html",
|
|
source_name=source_name.title(),
|
|
source_slug=source_name.lower(),
|
|
articles=articles_data,
|
|
pagination=pagination,
|
|
)
|
|
|
|
|
|
def validate_archive_path(archive_path: str) -> Optional[Path]:
|
|
"""Validate archive path is within ARCHIVE_DIR (prevents path traversal).
|
|
|
|
URL-decodes path first to block encoded traversal (..%2F).
|
|
|
|
Args:
|
|
archive_path: Requested path component
|
|
|
|
Returns:
|
|
Resolved Path if valid, None if traversal detected
|
|
"""
|
|
decoded = unquote(archive_path)
|
|
try:
|
|
archive_file = (ARCHIVE_DIR / decoded).resolve()
|
|
if not str(archive_file).startswith(str(ARCHIVE_DIR)):
|
|
logger.warning("Path traversal attempt blocked: %s", archive_path)
|
|
return None
|
|
return archive_file
|
|
except (ValueError, OSError):
|
|
return None
|
|
|
|
|
|
@app.route("/archive/<path:archive_path>")
|
|
@login_required
|
|
def serve_archive(archive_path):
|
|
"""Serve archived HTML file."""
|
|
archive_file = validate_archive_path(archive_path)
|
|
if archive_file and archive_file.exists():
|
|
return archive_file.read_text(encoding="utf-8")
|
|
return "Archive not found", 404
|
|
|
|
|
|
@app.route("/archive-file/<path:encoded_path>")
|
|
@login_required
|
|
def serve_archive_file(encoded_path):
|
|
"""Serve archived HTML file from encoded path."""
|
|
archive_file = validate_archive_path(encoded_path)
|
|
if archive_file and archive_file.exists():
|
|
return archive_file.read_text(encoding="utf-8")
|
|
return "Archive not found", 404
|
|
|
|
|
|
@app.route("/source/<slug>/article/<int:article_id>")
|
|
@login_required
|
|
def article(slug: str, article_id: int):
|
|
"""Individual article page."""
|
|
sources = get_all_sources()
|
|
source_name = None
|
|
for s in sources:
|
|
if s.lower() == slug.lower():
|
|
source_name = s
|
|
break
|
|
|
|
if not source_name:
|
|
return render_template(
|
|
"article_not_found.html", slug=slug, article_id=article_id
|
|
), 404
|
|
|
|
article = get_article(source_name, article_id)
|
|
if not article:
|
|
return render_template(
|
|
"article_not_found.html", slug=slug, article_id=article_id
|
|
), 404
|
|
|
|
article_data = {
|
|
"id": article_id,
|
|
"title": article.title or "Untitled",
|
|
"publish_date": article.publish_date or "",
|
|
"author": article.author or "",
|
|
"url": article.url or "",
|
|
"content_text": article.content_text or "",
|
|
"archive_file_path": article.archive_file_path or "",
|
|
}
|
|
|
|
return render_template(
|
|
"article.html",
|
|
source_name=source_name.title(),
|
|
source_slug=source_name.lower(),
|
|
article=article_data,
|
|
)
|
|
|
|
|
|
@app.route("/status")
|
|
@login_required
|
|
def status():
|
|
"""System status page."""
|
|
sources = get_all_sources()
|
|
rss_feeds = load_rss_feeds()
|
|
|
|
sources_info = []
|
|
disabled_sources = []
|
|
total_articles = 0
|
|
failed_jobs = 0
|
|
disabled_count = 0
|
|
|
|
for source_name in sources:
|
|
stats = get_source_stats(source_name)
|
|
|
|
source_info = {
|
|
"name": source_name.title(),
|
|
"slug": source_name,
|
|
"article_count": stats["total_articles"],
|
|
"last_archived": stats.get("last_archived"),
|
|
"status": "success" if stats["total_articles"] > 0 else "pending",
|
|
}
|
|
|
|
if source_name in rss_feeds:
|
|
feed_info = rss_feeds[source_name]
|
|
if feed_info.get("disabled", False):
|
|
source_info["disabled"] = True
|
|
disabled_count += 1
|
|
disabled_sources.append(source_info)
|
|
continue
|
|
|
|
sources_info.append(source_info)
|
|
total_articles += stats["total_articles"]
|
|
failed_jobs += stats["failed"]
|
|
|
|
sources_info.extend(disabled_sources)
|
|
|
|
return render_template(
|
|
"status.html",
|
|
sources=sources_info,
|
|
total_articles=total_articles,
|
|
failed_jobs=failed_jobs,
|
|
sources_monitored=len(sources) - disabled_count,
|
|
disabled_sources=disabled_count,
|
|
)
|
|
|
|
|
|
@app.route("/api/sources")
|
|
@login_required
|
|
def api_sources():
|
|
"""API endpoint for listing all sources."""
|
|
sources = get_all_sources()
|
|
rss_feeds = load_rss_feeds()
|
|
|
|
source_list = []
|
|
disabled_sources = []
|
|
|
|
for source_name in sources:
|
|
stats = get_source_stats(source_name)
|
|
|
|
source_info = {
|
|
"name": source_name.title(),
|
|
"slug": source_name,
|
|
"article_count": stats["total_articles"],
|
|
"last_archived": stats.get("last_archived"),
|
|
"status": "success" if stats["total_articles"] > 0 else "pending",
|
|
}
|
|
|
|
if source_name in rss_feeds:
|
|
feed_info = rss_feeds[source_name]
|
|
if feed_info.get("disabled", False):
|
|
source_info["disabled"] = True
|
|
source_info["disable_reason"] = feed_info.get(
|
|
"disable_reason", "No reason provided"
|
|
)
|
|
disabled_sources.append(source_info)
|
|
continue
|
|
|
|
source_list.append(source_info)
|
|
|
|
source_list.extend(disabled_sources)
|
|
|
|
return jsonify({"sources": source_list})
|
|
|
|
|
|
@app.route("/api/source/<slug>/articles")
|
|
@login_required
|
|
def api_articles(slug: str):
|
|
"""API endpoint for listing articles for a source."""
|
|
page = request.args.get("page", 1, type=int)
|
|
per_page = 50
|
|
|
|
sources = get_all_sources()
|
|
source_name = None
|
|
for s in sources:
|
|
if s.lower() == slug.lower():
|
|
source_name = s
|
|
break
|
|
|
|
if not source_name:
|
|
return jsonify({"error": "Source not found"}), 404
|
|
|
|
articles_list = get_articles_by_source(
|
|
source_name, limit=per_page, offset=(page - 1) * per_page
|
|
)
|
|
stats = get_source_stats(source_name)
|
|
total = stats["total_articles"]
|
|
|
|
pagination = get_pagination_info(total, page, per_page)
|
|
|
|
articles_data = []
|
|
for article in articles_list:
|
|
articles_data.append(
|
|
{
|
|
"id": getattr(article, "id", 0),
|
|
"title": article.title or "Untitled",
|
|
"date": article.publish_date or "",
|
|
"summary": article.content_text[:200] if article.content_text else "",
|
|
"url": f"/source/{source_name.lower()}/article/{getattr(article, 'id', 0)}",
|
|
}
|
|
)
|
|
|
|
return jsonify(
|
|
{
|
|
"source_name": source_name.title(),
|
|
"articles": articles_data,
|
|
"total": total,
|
|
"page": page,
|
|
"per_page": per_page,
|
|
"has_next": pagination["has_next"],
|
|
"has_prev": pagination["has_prev"],
|
|
}
|
|
)
|
|
|
|
|
|
@app.route("/api/status")
|
|
@login_required
|
|
def api_status():
|
|
"""API endpoint for system status."""
|
|
sources = get_all_sources()
|
|
|
|
sources_monitored = len(sources)
|
|
total_articles = 0
|
|
failed_jobs = 0
|
|
last_archive_run = None
|
|
|
|
for source_name in sources:
|
|
stats = get_source_stats(source_name)
|
|
total_articles += stats["total_articles"]
|
|
failed_jobs += stats["failed"]
|
|
|
|
if stats.get("last_archive_run"):
|
|
if last_archive_run is None or stats["last_archive_run"] > last_archive_run:
|
|
last_archive_run = stats["last_archive_run"]
|
|
|
|
return jsonify(
|
|
{
|
|
"status": "online",
|
|
"last_archive_run": last_archive_run,
|
|
"pending_jobs": 0,
|
|
"failed_jobs": failed_jobs,
|
|
"sources_monitored": sources_monitored,
|
|
"total_articles": total_articles,
|
|
}
|
|
)
|
|
|
|
|
|
@app.route("/rss")
|
|
def rss_feed():
|
|
"""RSS 2.0 endpoint for latest archived articles."""
|
|
limit = request.args.get("limit", 50, type=int)
|
|
|
|
articles = get_latest_articles(limit=limit)
|
|
|
|
rss_items = []
|
|
for article in articles:
|
|
if (
|
|
article.title
|
|
and article.content_text
|
|
and "Performing security verification" not in article.content_text
|
|
):
|
|
pub_date = None
|
|
if article.publish_date:
|
|
try:
|
|
dt = datetime.fromisoformat(
|
|
article.publish_date.replace("Z", "+00:00")
|
|
)
|
|
pub_date = dt.strftime("%a, %d %b %Y %H:%M:%S %z").strip()
|
|
except (ValueError, AttributeError):
|
|
try:
|
|
dt = datetime.fromisoformat(article.publish_date)
|
|
pub_date = dt.strftime("%a, %d %b %Y %H:%M:%S +0000")
|
|
except (ValueError, AttributeError):
|
|
pub_date = datetime.now(timezone.utc).strftime(
|
|
"%a, %d %b %Y %H:%M:%S +0000"
|
|
)
|
|
|
|
source_name = article.source_name or "unknown"
|
|
encoded_source = quote(source_name.lower())
|
|
item = {
|
|
"title": html.escape(article.title),
|
|
"link": f"{SERVER_URL}/source/{encoded_source}/article/{article.id}",
|
|
"pubDate": pub_date,
|
|
"description": html.escape(
|
|
article.content_text[:500] if article.content_text else ""
|
|
),
|
|
"guid": html.escape(article.url or f"article-{article.id}"),
|
|
}
|
|
if article.author:
|
|
item["author"] = html.escape(article.author)
|
|
rss_items.append(item)
|
|
|
|
rss_template = render_template(
|
|
"rss.xml",
|
|
title="NewsArchiver - Latest Articles",
|
|
link=SERVER_URL,
|
|
description="Latest archived news articles",
|
|
last_build_date=datetime.now(timezone.utc).strftime(
|
|
"%a, %d %b %Y %H:%M:%S +0000"
|
|
),
|
|
items=rss_items,
|
|
)
|
|
|
|
response = make_response(rss_template)
|
|
response.headers["Content-Type"] = "application/rss+xml; charset=utf-8"
|
|
return response
|
|
|
|
|
|
@app.route("/atom")
|
|
def atom_feed():
|
|
"""Atom 1.0 endpoint for latest archived articles."""
|
|
limit = request.args.get("limit", 50, type=int)
|
|
|
|
articles = get_latest_articles(limit=limit)
|
|
|
|
atom_entries = []
|
|
for article in articles:
|
|
if (
|
|
article.title
|
|
and article.content_text
|
|
and "Performing security verification" not in article.content_text
|
|
):
|
|
pub_date = None
|
|
if article.publish_date:
|
|
try:
|
|
dt = datetime.fromisoformat(
|
|
article.publish_date.replace("Z", "+00:00")
|
|
)
|
|
pub_date = dt.strftime("%Y-%m-%dT%H:%M:%S+00:00")
|
|
except (ValueError, AttributeError):
|
|
pub_date = datetime.now(timezone.utc).strftime(
|
|
"%Y-%m-%dT%H:%M:%S+00:00"
|
|
)
|
|
|
|
source_name = article.source_name or "unknown"
|
|
encoded_source = quote(source_name.lower())
|
|
entry = {
|
|
"title": html.escape(article.title),
|
|
"link": f"{SERVER_URL}/source/{encoded_source}/article/{article.id}",
|
|
"published": pub_date,
|
|
"summary": html.escape(
|
|
article.content_text[:500] if article.content_text else ""
|
|
),
|
|
"id": html.escape(article.url or f"article-{article.id}"),
|
|
}
|
|
if article.author:
|
|
entry["author"] = {"name": html.escape(article.author)}
|
|
atom_entries.append(entry)
|
|
|
|
atom_template = render_template(
|
|
"atom.xml",
|
|
title="NewsArchiver - Latest Articles",
|
|
link=SERVER_URL,
|
|
updated=datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S+00:00"),
|
|
entries=atom_entries,
|
|
)
|
|
|
|
response = make_response(atom_template)
|
|
response.headers["Content-Type"] = "application/atom+xml; charset=utf-8"
|
|
return response
|
|
|
|
|
|
def load_rss_feeds() -> dict:
|
|
"""Load RSS feeds configuration, resolving environment variables in API keys."""
|
|
global RSS_FEEDS
|
|
|
|
if RSS_FEEDS:
|
|
return RSS_FEEDS
|
|
|
|
if not RSS_FEEDS_PATH.exists():
|
|
logger.warning("RSS feeds file not found: %s", RSS_FEEDS_PATH)
|
|
return {}
|
|
|
|
try:
|
|
raw_content = RSS_FEEDS_PATH.read_text(encoding="utf-8")
|
|
for key, value in os.environ.items():
|
|
raw_content = raw_content.replace(f"${{{key}}}", value)
|
|
RSS_FEEDS = json.loads(raw_content)
|
|
return RSS_FEEDS
|
|
except Exception as e:
|
|
logger.error("Failed to load RSS feeds: %s", sanitize_error_message(str(e)))
|
|
return {}
|
|
|
|
|
|
if __name__ == "__main__":
|
|
logger.info("Starting web interface...")
|
|
|
|
if not DB_PATH.exists():
|
|
logger.info("Database not found, initializing...")
|
|
from storage_manager import initialize_storage
|
|
|
|
initialize_storage()
|
|
|
|
app.run(host="0.0.0.0", port=5000, debug=False)
|