JQC_features/app.py

import logging
import os
from datetime import datetime
from logging.handlers import RotatingFileHandler

import bleach
from flask import Flask, render_template
from flask_sqlalchemy import SQLAlchemy
from flask_wtf import CSRFProtect
from markupsafe import Markup
from werkzeug.middleware.proxy_fix import ProxyFix

from config import Config

db = SQLAlchemy()
csrf = CSRFProtect()


class Section(db.Model):
    __tablename__ = "section"
    id = db.Column(db.Integer, primary_key=True)
    num = db.Column(db.Integer, nullable=False, unique=True)
    title = db.Column(db.String(160), nullable=False)
    subtitle = db.Column(db.String(255))
    sort_order = db.Column(db.Integer, nullable=False, default=0)
    topics = db.relationship(
        "Topic",
        backref="section",
        order_by="Topic.sort_order",
        cascade="all, delete-orphan",
    )

    @property
    def published_topics(self):
        # Topics are already ordered by sort_order via the relationship.
        return [t for t in self.topics if t.is_published]


class Topic(db.Model):
    __tablename__ = "topic"
    id = db.Column(db.Integer, primary_key=True)
    section_id = db.Column(
        db.Integer, db.ForeignKey("section.id", ondelete="CASCADE"), nullable=False
    )
    slug = db.Column(db.String(80), nullable=False, unique=True)
    title = db.Column(db.String(200), nullable=False)
    body_html = db.Column(db.Text)
    link_url = db.Column(db.String(500))
    link_label = db.Column(db.String(120))
    media_type = db.Column(db.String(16), nullable=False, default="none")
    media_url = db.Column(db.String(500))
    media_caption = db.Column(db.String(255))
    sort_order = db.Column(db.Integer, nullable=False, default=0)
    is_published = db.Column(db.Boolean, nullable=False, default=True)

    @property
    def body(self):
        # Content is admin-authored and trusted; render as-is.
        return Markup(self.body_html or "")

    @property
    def is_youtube(self):
        u = (self.media_url or "").lower()
        return "youtube.com" in u or "youtu.be" in u


class AuditLog(db.Model):
    __tablename__ = "audit_log"
    id = db.Column(db.Integer, primary_key=True)
    actor = db.Column(db.String(80))
    action = db.Column(db.String(40), nullable=False)      # create / update / delete
    entity = db.Column(db.String(40), nullable=False)      # section / topic
    entity_id = db.Column(db.Integer)
    detail = db.Column(db.String(255))
    created_at = db.Column(db.DateTime, nullable=False, default=datetime.utcnow)


def log_action(actor, action, entity, entity_id=None, detail=None):
    """Record an audit row. MUST be called AFTER db.session.commit() of the
    change it describes, so a failed transaction never leaves an orphan log."""
    entry = AuditLog(
        actor=actor, action=action, entity=entity,
        entity_id=entity_id, detail=detail,
    )
    db.session.add(entry)
    db.session.commit()


# Tags/attributes the rich-text editor (Quill) can emit. Everything else is
# stripped on save so a WYSIWYG paste can't inject markup into the public page.
ALLOWED_TAGS = [
    "p", "br", "strong", "b", "em", "i", "u", "s", "strike",
    "ul", "ol", "li", "a", "h2", "h3", "blockquote",
    # images (inserted via the /admin/upload endpoint or a URL)
    "img",
    # tables (Quill 2 built-in table module)
    "table", "thead", "tbody", "tr", "td", "th", "col", "colgroup",
]
ALLOWED_ATTRS = {
    "a": ["href", "title", "target", "rel"],
    "img": ["src", "alt", "width", "height"],
    # Quill 2 tags cells/rows with data-row; keep the standard span attrs too.
    "table": ["class"],
    "td": ["data-row", "colspan", "rowspan"],
    "th": ["data-row", "colspan", "rowspan"],
    "tr": ["data-row"],
    "col": ["width"],
}


def sanitize_html(raw):
    """Clean editor HTML against the allowlist. Returns None for empty content
    so blank bodies stay NULL. bleach also restricts URL protocols to
    http/https/mailto for both links and images, blocking javascript: and
    data: URLs (uploaded images are served from a relative /static path)."""
    if not raw:
        return None
    cleaned = bleach.clean(
        raw, tags=ALLOWED_TAGS, attributes=ALLOWED_ATTRS, strip=True
    ).strip()
    # Quill leaves an empty paragraph for a blank editor.
    if cleaned in ("", "<p></p>", "<p><br></p>"):
        return None
    return cleaned


def _configure_auth_logger(app):
    """A dedicated 'jqc.auth' logger writing one line per login attempt to a
    file fail2ban watches. Kept separate from the app log so the filter regex
    stays tight and rotation is self-contained (no logrotate needed)."""
    log_path = app.config.get("AUTH_LOG_PATH") or os.path.join(
        os.path.dirname(os.path.abspath(__file__)), "logs", "auth.log"
    )
    os.makedirs(os.path.dirname(log_path), exist_ok=True)

    auth_log = logging.getLogger("jqc.auth")
    auth_log.setLevel(logging.INFO)
    auth_log.propagate = False
    # Guard against duplicate handlers if create_app runs more than once.
    if not any(isinstance(h, RotatingFileHandler) for h in auth_log.handlers):
        handler = RotatingFileHandler(
            log_path, maxBytes=1_000_000, backupCount=5, encoding="utf-8"
        )
        handler.setFormatter(
            logging.Formatter("%(asctime)s %(name)s %(levelname)s %(message)s")
        )
        auth_log.addHandler(handler)
    return auth_log


def create_app():
    app = Flask(__name__)
    app.config.from_object(Config)

    # Behind nginx: trust ONE proxy hop so request.remote_addr / scheme reflect
    # the real client (nginx sets X-Forwarded-For / -Proto). gunicorn binds
    # 127.0.0.1 only, so these headers can't be spoofed from outside.
    app.wsgi_app = ProxyFix(app.wsgi_app, x_for=1, x_proto=1, x_host=1)

    db.init_app(app)
    csrf.init_app(app)
    _configure_auth_logger(app)

    # Deferred import avoids a circular import: admin.py imports the models and
    # log_action defined above, which are ready by the time create_app() runs.
    from admin import admin_bp
    app.register_blueprint(admin_bp)

    @app.route("/")
    def index():
        all_sections = (
            Section.query.order_by(Section.sort_order, Section.num).all()
        )
        # Hide sections whose topics are all drafts (or which have no topics).
        sections = [s for s in all_sections if s.published_topics]
        return render_template(
            "index.html",
            sections=sections,
            demo_url=app.config["DEMO_CONTACT_URL"],
        )

    @app.route("/healthz")
    def healthz():
        return {"status": "ok"}

    return app


app = create_app()


if __name__ == "__main__":
    app.run(host="127.0.0.1", port=8000, debug=True)