441 lines
16 KiB
Python
441 lines
16 KiB
Python
import logging
|
|
import os
|
|
import re
|
|
import threading
|
|
import time
|
|
from collections import defaultdict
|
|
from datetime import date, datetime, timedelta
|
|
from logging.handlers import RotatingFileHandler
|
|
|
|
import bleach
|
|
from bleach.css_sanitizer import CSSSanitizer
|
|
from flask import Flask, flash, redirect, render_template, request, url_for
|
|
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
|
|
from mailer import send_email_async
|
|
|
|
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 DemoRequest(db.Model):
|
|
"""A demo appointment requested from the public page. `preferred_date` /
|
|
`preferred_time` are what the customer picked (no timezone conversion —
|
|
they're read back as the site's advertised local business hours)."""
|
|
__tablename__ = "demo_request"
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
name = db.Column(db.String(120), nullable=False)
|
|
company = db.Column(db.String(160))
|
|
email = db.Column(db.String(200), nullable=False)
|
|
phone = db.Column(db.String(40))
|
|
preferred_date = db.Column(db.Date, nullable=False)
|
|
preferred_time = db.Column(db.String(5), nullable=False) # 'HH:MM'
|
|
message = db.Column(db.Text)
|
|
status = db.Column(db.String(16), nullable=False, default="new")
|
|
source_ip = db.Column(db.String(45))
|
|
created_at = db.Column(db.DateTime, nullable=False, default=datetime.utcnow)
|
|
|
|
@property
|
|
def when(self):
|
|
return f"{self.preferred_date:%a %b %d, %Y} at {self.preferred_time}"
|
|
|
|
|
|
DEMO_STATUSES = ("new", "scheduled", "done", "cancelled")
|
|
|
|
|
|
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-table-better: resizable cols, aligned/styled cells)
|
|
"table", "thead", "tbody", "tr", "td", "th", "col", "colgroup",
|
|
]
|
|
# quill-table-better carries column widths, borders and background as inline
|
|
# `style`, plus data-* bookkeeping attributes. Cell alignment is Quill's
|
|
# `ql-align-*` CSS class on the cell's <p> block, and quill-table-better also
|
|
# tags blots with `ql-table-block`/`table-th-block` classes — all of which must
|
|
# survive so the table (and its alignment) round-trips back into the editor.
|
|
_TABLE_CELL_ATTRS = [
|
|
"data-row", "data-cell", "data-class", "colspan", "rowspan",
|
|
"width", "height", "style", "class",
|
|
]
|
|
ALLOWED_ATTRS = {
|
|
"a": ["href", "title", "target", "rel"],
|
|
"img": ["src", "alt", "width", "height"],
|
|
"p": ["style", "class", "data-cell", "data-row"],
|
|
"table": ["class", "style", "align", "width", "height", "data-class"],
|
|
"colgroup": ["style", "class"],
|
|
"col": ["width", "span", "style", "class"],
|
|
"tr": ["data-row", "style", "class"],
|
|
"td": _TABLE_CELL_ATTRS,
|
|
"th": _TABLE_CELL_ATTRS,
|
|
}
|
|
# Only these CSS properties survive on a `style` attribute — enough for the
|
|
# table editor's sizing/alignment/borders, nothing that can smuggle script.
|
|
ALLOWED_CSS_PROPS = [
|
|
"width", "height", "min-width", "padding", "text-align", "vertical-align",
|
|
"background-color", "border", "border-style", "border-color",
|
|
"border-width", "border-collapse",
|
|
]
|
|
_css_sanitizer = CSSSanitizer(allowed_css_properties=ALLOWED_CSS_PROPS)
|
|
|
|
|
|
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). Inline
|
|
`style` is filtered to the ALLOWED_CSS_PROPS allowlist via CSSSanitizer."""
|
|
if not raw:
|
|
return None
|
|
cleaned = bleach.clean(
|
|
raw, tags=ALLOWED_TAGS, attributes=ALLOWED_ATTRS, strip=True,
|
|
css_sanitizer=_css_sanitizer,
|
|
).strip()
|
|
# Quill leaves an empty paragraph for a blank editor.
|
|
if cleaned in ("", "<p></p>", "<p><br></p>"):
|
|
return None
|
|
return cleaned
|
|
|
|
|
|
# ---------------------------------------------------------------- demo form
|
|
# Loose sanity check only — the real proof an address works is the confirmation
|
|
# mail the customer receives. Rejecting exotic-but-valid addresses is worse than
|
|
# accepting a typo the owner can see and follow up on.
|
|
EMAIL_RE = re.compile(r"^[^@\s]+@[^@\s.]+(\.[^@\s.]+)+$")
|
|
|
|
# In-memory per-IP throttle. Per gunicorn worker (not shared), which is fine:
|
|
# it exists to stop a bored visitor spamming the form, not a distributed flood.
|
|
# nginx limit_req is the right tool if that ever matters.
|
|
_demo_hits = defaultdict(list)
|
|
_demo_hits_lock = threading.Lock()
|
|
|
|
|
|
def demo_slots(config):
|
|
"""Bookable clock times as 'HH:MM', e.g. 08:00, 08:30, ... 16:30."""
|
|
start = max(0, min(23, config.get("DEMO_HOUR_START", 8)))
|
|
end = max(start + 1, min(24, config.get("DEMO_HOUR_END", 17)))
|
|
step = max(5, config.get("DEMO_SLOT_MINUTES", 30))
|
|
slots, minute = [], start * 60
|
|
while minute < end * 60:
|
|
slots.append(f"{minute // 60:02d}:{minute % 60:02d}")
|
|
minute += step
|
|
return slots
|
|
|
|
|
|
def _rate_limited(ip, limit):
|
|
"""True if `ip` has already submitted `limit` requests in the last hour."""
|
|
if limit <= 0:
|
|
return False
|
|
now = time.time()
|
|
with _demo_hits_lock:
|
|
hits = [t for t in _demo_hits[ip] if now - t < 3600]
|
|
if len(hits) >= limit:
|
|
_demo_hits[ip] = hits
|
|
return True
|
|
hits.append(now)
|
|
_demo_hits[ip] = hits
|
|
return False
|
|
|
|
|
|
def _notify_demo(app, req):
|
|
"""Mail the owner about a new appointment, and (optionally) confirm to the
|
|
customer. Both are fire-and-forget: the row is already committed."""
|
|
cfg = app.config
|
|
admin_link = request.url_root.rstrip("/") + url_for("admin.demos")
|
|
owner_body = (
|
|
"A new demo appointment was requested from the JQC features site.\n\n"
|
|
f"When: {req.when} ({cfg['DEMO_TIMEZONE_LABEL']})\n"
|
|
f"Name: {req.name}\n"
|
|
f"Company: {req.company or '—'}\n"
|
|
f"Email: {req.email}\n"
|
|
f"Phone: {req.phone or '—'}\n\n"
|
|
f"Message:\n{req.message or '—'}\n\n"
|
|
f"Manage requests: {admin_link}\n"
|
|
)
|
|
send_email_async(
|
|
cfg, cfg.get("DEMO_NOTIFY_EMAIL"),
|
|
f"Demo request — {req.name} · {req.when}",
|
|
owner_body,
|
|
reply_to=req.email, reply_name=req.name,
|
|
)
|
|
|
|
if cfg.get("DEMO_CONFIRM_CUSTOMER"):
|
|
send_email_async(
|
|
cfg, req.email,
|
|
"We received your demo request",
|
|
(
|
|
f"Hi {req.name},\n\n"
|
|
"Thanks for your interest in JQC — LT Services' Janitorial "
|
|
"Quality Control program.\n\n"
|
|
f"You asked for a demonstration on {req.when} "
|
|
f"({cfg['DEMO_TIMEZONE_LABEL']}). We'll confirm that time by "
|
|
"email or phone shortly; if it turns out not to work on our "
|
|
"side we'll suggest the nearest alternative.\n\n"
|
|
"— LT Services Inc.\n"
|
|
),
|
|
)
|
|
|
|
|
|
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("/demo", methods=["GET", "POST"])
|
|
def demo():
|
|
"""Public demo-appointment form. On success the request is stored and
|
|
the owner is notified by mail (best-effort, on a background thread)."""
|
|
cfg = app.config
|
|
slots = demo_slots(cfg)
|
|
today = date.today()
|
|
max_date = today + timedelta(days=cfg["DEMO_MAX_DAYS_AHEAD"])
|
|
form = {}
|
|
|
|
if request.method == "POST":
|
|
f = request.form
|
|
form = {k: f.get(k, "").strip() for k in (
|
|
"name", "company", "email", "phone", "preferred_date",
|
|
"preferred_time", "message",
|
|
)}
|
|
errors = []
|
|
|
|
# Honeypot: a hidden field only a bot fills in. Pretend success so
|
|
# the bot doesn't learn to work around it.
|
|
if f.get("website", "").strip():
|
|
return redirect(url_for("demo_thanks"))
|
|
|
|
client_ip = request.remote_addr or "-"
|
|
if _rate_limited(client_ip, cfg["DEMO_RATE_LIMIT"]):
|
|
errors.append(
|
|
"Too many requests from this connection. Please try again "
|
|
"later or email us directly."
|
|
)
|
|
|
|
if not form["name"]:
|
|
errors.append("Please tell us your name.")
|
|
if not EMAIL_RE.match(form["email"] or ""):
|
|
errors.append("Please enter a valid email address.")
|
|
|
|
try:
|
|
wanted = datetime.strptime(form["preferred_date"], "%Y-%m-%d").date()
|
|
except ValueError:
|
|
wanted = None
|
|
errors.append("Please pick a date for the demonstration.")
|
|
if wanted:
|
|
if wanted < today:
|
|
errors.append("Please pick a date that hasn't passed yet.")
|
|
elif wanted > max_date:
|
|
errors.append(
|
|
f"Please pick a date within the next "
|
|
f"{cfg['DEMO_MAX_DAYS_AHEAD']} days."
|
|
)
|
|
elif wanted.weekday() >= 5:
|
|
errors.append("Demonstrations run Monday through Friday.")
|
|
|
|
if form["preferred_time"] not in slots:
|
|
errors.append("Please pick an available time.")
|
|
|
|
# One appointment per slot. Cancelled ones free the slot again.
|
|
if not errors and DemoRequest.query.filter(
|
|
DemoRequest.preferred_date == wanted,
|
|
DemoRequest.preferred_time == form["preferred_time"],
|
|
DemoRequest.status != "cancelled",
|
|
).first():
|
|
errors.append("That time was just taken — please pick another.")
|
|
|
|
if errors:
|
|
for msg in errors:
|
|
flash(msg, "error")
|
|
else:
|
|
req = DemoRequest(
|
|
name=form["name"][:120],
|
|
company=form["company"][:160] or None,
|
|
email=form["email"][:200],
|
|
phone=form["phone"][:40] or None,
|
|
preferred_date=wanted,
|
|
preferred_time=form["preferred_time"],
|
|
message=form["message"][:2000] or None,
|
|
status="new",
|
|
source_ip=client_ip[:45],
|
|
)
|
|
db.session.add(req)
|
|
db.session.commit()
|
|
log_action(
|
|
"public", "create", "demo", req.id,
|
|
f"{req.name} · {req.when}",
|
|
)
|
|
_notify_demo(app, req)
|
|
return redirect(url_for("demo_thanks"))
|
|
|
|
return render_template(
|
|
"demo.html",
|
|
slots=slots,
|
|
form=form,
|
|
min_date=today.isoformat(),
|
|
max_date=max_date.isoformat(),
|
|
tz_label=cfg["DEMO_TIMEZONE_LABEL"],
|
|
demo_url=cfg["DEMO_CONTACT_URL"],
|
|
)
|
|
|
|
@app.route("/demo/thanks")
|
|
def demo_thanks():
|
|
return render_template("demo_thanks.html", demo_url=app.config["DEMO_CONTACT_URL"])
|
|
|
|
@app.route("/healthz")
|
|
def healthz():
|
|
return {"status": "ok"}
|
|
|
|
@app.context_processor
|
|
def inject_asset_url():
|
|
"""`asset_url('css/style.css')` → the static URL with a `?v=<mtime>`
|
|
cache-buster, so an edited CSS/JS file is always re-fetched instead of
|
|
being served stale from the browser/proxy cache."""
|
|
from flask import url_for
|
|
|
|
def asset_url(filename):
|
|
try:
|
|
version = int(os.path.getmtime(
|
|
os.path.join(app.static_folder, filename)))
|
|
except OSError:
|
|
version = 0
|
|
return url_for("static", filename=filename, v=version)
|
|
|
|
return {"asset_url": asset_url}
|
|
|
|
return app
|
|
|
|
|
|
app = create_app()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
app.run(host="127.0.0.1", port=8000, debug=True)
|