dynette/app.py

import hmac
import base64
import os
import re
import yaml
import bcrypt

from flask import Flask, jsonify, request
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address

DOMAIN_REGEX = re.compile(
    r"^([a-z0-9]{1}([a-z0-9\-]*[a-z0-9])*)(\.[a-z0-9]{1}([a-z0-9\-]*[a-z0-9])*)*(\.[a-z]{1}([a-z0-9\-]*[a-z0-9])*)$"
)

app = Flask(__name__)
app.config.from_file("config.yml", load=yaml.safe_load)
limiter = Limiter(
    get_remote_address,
    app=app,
    default_limits=["50 per hour"],
    storage_uri="memory://",
)

assert os.path.isdir(
    app.config["DB_FOLDER"]
), "You should create the DB folder declared in the config"


def _validate_domain(domain):

    if not DOMAIN_REGEX.match(domain):
        return {"error": f"This is not a valid domain: {domain}"}, 400

    if (
        len(domain.split(".")) != 3
        or domain.split(".", 1)[-1] not in app.config["DOMAINS"]
    ):
        return {"error": f"This subdomain is not handled by this dynette server."}, 400


def _is_available(domain):

    return not os.path.exists(f"{app.config['DB_FOLDER']}/{domain}.key")


@app.route("/")
@limiter.exempt
def home():
    return "Wanna play the dynette?"


@app.route("/domains")
@limiter.exempt
def domains():
    return jsonify(app.config["DOMAINS"]), 200


@app.route("/test/<domain>")
@limiter.limit("3 per minute")
def availability(domain):

    error = _validate_domain(domain)
    if error:
        return error

    if _is_available(domain):
        return f"Domain {domain} is available", 200
    else:
        return {"error": f"Subdomain already taken: {domain}"}, 409


@app.route("/key/<key>", methods=["POST"])
@limiter.limit("5 per hour")
def register(key):

    try:
        key = base64.b64decode(key).decode()
    except Exception as e:
        return {"error": "Key format is invalid"}, 400
    else:
        if len(key) != 89:
            return {"error": "Key format is invalid"}, 400

    try:
        data = request.get_json(force=True)
        assert isinstance(data, dict)
        subdomain = data.get("subdomain")
        assert isinstance(subdomain, str)
    except Exception:
        return {"error": "Invalid request"}, 400

    error = _validate_domain(subdomain)
    if error:
        return error

    if not _is_available(subdomain):
        return {"error": f"Subdomain already taken: {subdomain}"}, 409

    recovery_password = data.get("recovery_password")
    if recovery_password and isinstance(recovery_password, str):
        if len(recovery_password) < 8:
            return {"error": "Recovery password too short"}, 409
        if len(recovery_password) > 1024:
            return {"error": "Recovery password too long"}, 409

        r_init = recovery_password
        recovery_password = bcrypt.hashpw(
            password=recovery_password.encode(), salt=bcrypt.gensalt(14)
        )
        recovery_password = base64.b64encode(recovery_password).decode()

    with open(f"{app.config['DB_FOLDER']}/{subdomain}.key", "w") as f:
        f.write(key)

    if recovery_password:
        with open(f"{app.config['DB_FOLDER']}/{subdomain}.recovery_password", "w") as f:
            f.write(recovery_password)

    return "OK", 201


@app.route("/domains/<subdomain>", methods=["DELETE"])
@limiter.limit("5 per hour")
def delete_using_recovery_password_or_key(subdomain):

    try:
        assert isinstance(subdomain, str)
        data = request.get_json(force=True)
        assert isinstance(data, dict)
        recovery_password = data.get("recovery_password")
        key = data.get("key")
        assert (recovery_password and isinstance(recovery_password, str)) or (
            key and isinstance(key, str)
        )
        if key:
            key = base64.b64decode(key).decode()
    except Exception:
        return {"error": "Invalid request"}, 400

    error = _validate_domain(subdomain)
    if error:
        return error

    if _is_available(subdomain):
        return {"error": "Subdomain already deleted"}, 409

    if key:
        with open(f"{app.config['DB_FOLDER']}/{subdomain}.key") as f:
            if not hmac.compare_digest(key, f.read()):
                return "Access denied", 403
    if recovery_password:
        if not os.path.exists(
            f"{app.config['DB_FOLDER']}/{subdomain}.recovery_password"
        ):
            return "Access denied", 403
        with open(f"{app.config['DB_FOLDER']}/{subdomain}.recovery_password") as f:
            hashed = base64.b64decode(f.read())

        if not bcrypt.checkpw(recovery_password.encode(), hashed):
            return "Access denied", 403

    if os.path.exists(f"{app.config['DB_FOLDER']}/{subdomain}.key"):
        os.remove(f"{app.config['DB_FOLDER']}/{subdomain}.key")
    if os.path.exists(f"{app.config['DB_FOLDER']}/{subdomain}.recovery_password"):
        os.remove(f"{app.config['DB_FOLDER']}/{subdomain}.recovery_password")

    return "OK", 200