path: root/app.py

import json
from hashlib import scrypt
from os import urandom
from hmac import compare_digest
import sqlite3
from colorsys import rgb_to_hsv

from flask import (
    Flask,
    render_template,
    request,
    redirect,
    url_for,
    session,
    g,
    flash,
)
from werkzeug.exceptions import HTTPException, NotImplemented, BadRequest
import requests

from captcha import Captcha
from presets import GUEST_CAPTCHA, GUIDE_CAPTCHAS

SCRYPT_PARAMS = {"n": 16384, "r": 8, "p": 1}
HEX_CHARS = set("0123456789ABCDEF")


def get_db():
    if "db" not in g:
        g.db = sqlite3.connect("akasha.db", detect_types=sqlite3.PARSE_DECLTYPES)
        g.db.row_factory = sqlite3.Row
    return g.db


def teardown(e=None):
    if (db := g.pop("db", None)) is not None:
        db.close()


def has_captcha(captcha_id):
    return captcha_id in session.get("captcha", [])


def add_captcha(captcha_id):
    if "captcha" not in session:
        session["captcha"] = []
    if captcha_id not in session["captcha"]:
        session["captcha"].append(captcha_id)
        session.modified = True


def make_app():
    global LIGHT_URL
    app = Flask(__name__)
    with sqlite3.connect("akasha.db", detect_types=sqlite3.PARSE_DECLTYPES) as temp_db:
        cur = temp_db.cursor()
        cur.execute("SELECT value FROM config WHERE key='fsk'")
        app.config["SECRET_KEY"] = cur.fetchone()[0]
        cur.execute("SELECT value FROM config WHERE key='light'")
        LIGHT_URL = f"http://192.168.1.10/api/{cur.fetchone()[0]}"
    app.teardown_appcontext(teardown)
    return app


app = make_app()


@app.context_processor
def context_processor():
    return dict(has_captcha=has_captcha)


@app.before_request
def before_request():
    if not request.endpoint or request.endpoint == "static":
        return
    db = get_db()
    cur = db.cursor()
    cur.execute("SELECT public FROM permit WHERE endpoint=?", (request.endpoint,))
    public = cur.fetchone()
    if public is None:
        raise NotImplemented
    match public[0]:
        case 2:
            return
        case 1:
            if "username" in session:
                return
            flash("You must be logged in to access this content.")
            return redirect(url_for("guest_login"))
        case 0:
            if "username" in session:
                cur = db.cursor()
                cur.execute(
                    "SELECT 1 FROM permit_override WHERE username=? AND endpoint=?",
                    (session["username"], request.endpoint),
                )
                if cur.fetchone():
                    return
                flash("This content requires additional permissions to view.")
            else:
                flash("You must be logged in to access this content.")
            return redirect(url_for("login"))
        case _:
            raise NotImplemented


@app.errorhandler(HTTPException)
def handle_bad_request(error):
    if error.code == 501:
        if "username" in session:
            flash("This content requires additional permissions to view.")
        else:
            flash("You must be logged in to access this content.")
        return redirect(url_for("login"))
    return render_template("error.html", error=error), error.code


@app.route("/")
def index():
    db = get_db()
    cur = db.cursor()
    cur.execute("SELECT * FROM news ORDER BY id DESC LIMIT 1")
    results = cur.fetchone()
    return render_template("index.html", news=results)


@app.route("/captcha")
def captcha():
    size = request.args.get("size", "")
    label = request.args.get("label", "")
    lock = request.args.get("lock", "")
    answer = request.args.get("answer", "")
    captcha = Captcha.safe_validate(size, label, lock, answer)
    if captcha:
        add_captcha(captcha.id)
        return {"valid": True}, 200
    return {"valid": False}, 400


@app.route("/login", methods=["GET", "POST"])
def login():
    db = get_db()
    if request.method == "POST":
        username = request.form.get("username")
        password = request.form.get("password")
        if not username or not password:
            flash("Error: Empty username or password.")
            return redirect(url_for("login"))
        if request.form.get("register") == "1":
            cur = db.cursor()
            cur.execute("SELECT 1 FROM users WHERE username=?", (username,))
            if cur.fetchone():
                flash("Error: User already exists.")
                return redirect(url_for("login"))
            salt = urandom(32)
            hash = scrypt(password.encode("utf-8"), salt=salt, **SCRYPT_PARAMS)
            cur.execute(
                "INSERT INTO users (username, hash, salt) VALUES (?, ?, ?)",
                (username, hash, salt),
            )
            flash("Success: User registered.")
            db.commit()
        else:
            cur = db.cursor()
            cur.execute("SELECT hash, salt FROM users WHERE username=?", (username,))
            results = cur.fetchone()
            if not results:
                flash("Error: Invalid username or password.")
                return redirect(url_for("login"))
            expected_hash, salt = results
            actual_hash = scrypt(password.encode("utf-8"), salt=salt, **SCRYPT_PARAMS)
            if not compare_digest(expected_hash, actual_hash):
                flash("Error: Invalid username or password.")
                return redirect(url_for("login"))
            session["username"] = username
            flash(f"Success: Logged in as {username}.")
        return redirect(url_for("index"))
    return render_template("login.html")


@app.route("/guest")
def guest_login():
    if has_captcha(GUEST_CAPTCHA.id) and "username" not in session:
        session["username"] = "guest"
        flash(f"Success: Logged in as guest.")
        return redirect(url_for("index"))
    return render_template("guest.html", captcha=GUEST_CAPTCHA)


@app.route("/eguide")
def eguide():
    return render_template("eguide.html", captchas=GUIDE_CAPTCHAS)


@app.route("/logout")
def logout():
    if "username" in session:
        del session["username"]
    if "captcha" in session:
        del session["captcha"]
    flash("Success: You have been logged out.")
    return redirect(url_for("index"))


@app.route("/hyde")
def hydematch():
    db = get_db()
    cur = db.cursor()
    cur.execute("SELECT * FROM matches ORDER BY id DESC")
    results = cur.fetchall()
    return render_template("hyde.html", matches=results)


@app.route("/news")
def news():
    db = get_db()
    cur = db.cursor()
    cur.execute("SELECT * FROM news ORDER BY id DESC")
    results = cur.fetchall()
    return render_template("news.html", news=results)


@app.route("/mgmt", methods=["GET", "POST"])
def management():
    if request.method == "POST":
        try:
            lid_key = request.form.get("lid", "custom")
            if lid_key == "custom":
                lid = int(request.form.get("custom", "5"))
            else:
                lid = int(lid_key)
        except ValueError:
            raise BadRequest
        set_key = request.form.get("set")
        match set_key:
            case "off":
                tool_disable_light(lid)
            case "on":
                tool_enable_light(lid)
            case "color":
                tool_color_light(lid, request.form.get("color", "#FF00AA"))
            case _:
                raise BadRequest
    return render_template("mgmt.html")


def tool_disable_light(light: int):
    requests.put(
        f"{LIGHT_URL}/lights/{light}/state",
        json={"on": False},
        timeout=5,
    )


def tool_enable_light(light: int):
    requests.put(
        f"{LIGHT_URL}/lights/{light}/state",
        json={"on": True, "ct": 366},
        timeout=5,
    )


def tool_color_light(light: int, color: str):
    if color[0] == "#":
        color = color[1:]
    color = color.upper()
    if len(color) != 6 or not all(c in HEX_CHARS for c in color):
        color = "FF00AA"
    rgb = [int(color[i : i + 2], 16) / 255 for i in range(0, 6, 2)]
    hsv = rgb_to_hsv(*rgb)
    requests.put(
        f"{LIGHT_URL}/lights/{light}/state",
        json={
            "on": True,
            "hue": int(hsv[0] * 65535),
            "sat": int(hsv[1] * 254),
            "bri": int(hsv[2] * 254),
        },
        timeout=5,
    )


@app.route("/permits")
def permits():
    raise NotImplemented


@app.route("/radio")
def radio():
    raise NotImplemented


@app.route("/absidy")
def absidy():
    raise NotImplemented