147 lines
4.8 KiB
Python
147 lines
4.8 KiB
Python
import logging
|
|
import os
|
|
from datetime import datetime
|
|
from logging.handlers import RotatingFileHandler
|
|
|
|
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",
|
|
)
|
|
|
|
|
|
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)
|
|
|
|
@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()
|
|
|
|
|
|
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():
|
|
sections = (
|
|
Section.query.order_by(Section.sort_order, Section.num).all()
|
|
)
|
|
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)
|