inventory-app/inventory-app/main.py

#!/usr/bin/env python3
import threading
import urllib
from configparser import ConfigParser
from datetime import datetime, timedelta
from functools import wraps
from http.server import BaseHTTPRequestHandler, HTTPServer

import bleach
import jinja2
import markdown
import segno
from flask import Flask, abort, g, redirect, request
from jinja2 import Environment, FileSystemLoader
from markupsafe import Markup
from prometheus_client import Gauge, CollectorRegistry, generate_latest
from prometheus_flask_exporter import PrometheusMetrics
from pymongo import MongoClient
from wtforms import (
    SelectMultipleField,
    ValidationError,
    widgets,
)

import const
from common import devenv, format_name, get_users, User
from inventory import page_inventory
from oidc import page_oidc, login_required, read_user
from doorboy import page_doorboy
from api import page_api

def check_foreign_key_format(item):
    owner = item.get("inventory", {}).get("owner", {})
    user = item.get("inventory", {}).get("user", {})
    if owner.get("foreign_id", False):
        return not bool(owner.get("username"))
    if user.get("foreign_id", False):
        return not bool(user.get("username"))

def render_user_display_name(username):
    if not username:
        return ""
    display_name = g.users_lookup.get(username, User()).display_name
    return display_name or username

def render_markdown(text):
    if not text:
        return ""
    cleaned = bleach.clean(text)
    md = markdown.markdown(cleaned)
    return Markup(md)

def render_timeago(dt):
    if not dt:
        return ""
    return Markup("<time class=\"timeago\" datetime=\"%s+0000\">%s</time>" % (dt, dt))

def render_last_seen(machine, stale_minutes=30):
    stale_timestamp = datetime.utcnow() - timedelta(minutes=stale_minutes)
    if machine and machine.get("last_seen"):
        if machine.get("last_seen") < stale_timestamp:
            return Markup("opacity: 25%;")

def render_owner_link(item):
    username = item.get("inventory", {}).get("owner", {}).get("username", False)
    if username:
        return Markup("<a href=\"/m/user/%s\">%s</a>" % (username, render_user_display_name(username)))
    else:
        return Markup("<form method=\"post\" action=\"/m/inventory/%s/claim\"><button type=\"submit\" class=\"waves-effect waves-light btn\">Mine!</button></form>" % (item["_id"]))

def render_user_link(item):
    username = item.get("inventory", {}).get("user", {}).get("username", False)
    if username:
        return Markup("<a href=\"/m/user/%s\">%s</a>" % (username, render_user_display_name(username)))
    elif item.get("inventory", {}).get("usable"):
        return Markup("<form method=\"post\" action=\"/m/inventory/%s/use\"><button type=\"submit\" class=\"waves-effect waves-light btn\">I use it now!</button></form>" % (item["_id"]))
    else:
        return ""

def is_list(value):
    return isinstance(value, list)

def render_qr_code(slug):
    qr = segno.make(str("http://k6.ee/" + slug), micro=False, mode='byte', mask=3, error='L', boost_error=False)
    return Markup(qr.svg_inline(light='black', dark='yellow', scale=3, border=1))

jinja2.filters.FILTERS['format_name'] = format_name
jinja2.filters.FILTERS['markdown'] = render_markdown
jinja2.filters.FILTERS['timeago'] = render_timeago
jinja2.filters.FILTERS['last_seen'] = render_last_seen
jinja2.filters.FILTERS['owner_link'] = render_owner_link
jinja2.filters.FILTERS['user_link'] = render_user_link
jinja2.filters.FILTERS['is_list'] = is_list
jinja2.filters.FILTERS['quote_plus'] = lambda u: urllib.parse.quote_plus(u)
jinja2.filters.FILTERS['check_foreign_key_format'] = check_foreign_key_format
jinja2.filters.FILTERS['display_name'] = render_user_display_name
jinja2.filters.FILTERS['qr_code'] = render_qr_code

env = Environment(loader=FileSystemLoader('templates/'))


class ReverseProxied(object):
    def __init__(self, app):
        self.app = app

    def __call__(self, environ, start_response):
        scheme = environ.get('HTTP_X_FORWARDED_PROTO')
        if scheme and scheme in ['http', 'https']:
            environ['wsgi.url_scheme'] = scheme
        return self.app(environ, start_response)

app = Flask(__name__)
app.wsgi_app = ReverseProxied(app.wsgi_app)
app.register_blueprint(page_inventory)
app.register_blueprint(page_oidc)
app.register_blueprint(page_api)
app.register_blueprint(page_doorboy)
metrics = PrometheusMetrics(app, group_by="path")

app.config['SECRET_KEY'] = const.SECRET_KEY

mongoclient = MongoClient(const.MONGO_URI)
mongodb = mongoclient.get_default_database()
mongodb.member.create_index("ad.username", sparse=True, unique=True)
mongodb.inventory.create_index("shortener.slug", sparse=True, unique=True)
mongodb.inventory.create_index("token.uid_hash", sparse=True, unique=True)
#mongodb.inventory.create_index("token.uid_hash", unique=True)

CATEGORY_COLORS = (
  ('membership-fee', '#acc236'),
  ('rent', '#f53794'),
  ('accounting-fees', '#f67019'),
  ('office-supplies', '#58595b'),
  ('server-room', '#166a8f'),
  ('training', '#00a950'),
  ('membership-fee-company', '#4dc9f6'),
  ('snack-machine', '#8549ba'),
  ('other', '#537bc4'),
)

INCOME_CATEGORIES = "membership-fee", "workshop"

cp = ConfigParser()
cp.read("/config/provision.conf")

PROVISION_CONFIGS = {}
for section in cp.sections():
    if not cp.getint(section, "enabled"):
        continue
    PROVISION_CONFIGS[section] = {
        "entrypoint": cp.get(section, "entrypoint"),
        "admin": cp.get(section, "admin"),
        "type": cp.get(section, "type"),
        "comment": cp.get(section, "comment"),
        "username": cp.get(section, "username"),
        "password": cp.get(section, "password"),
        "port": cp.get(section, "port"),
    }


def dev_only(f):
    @wraps(f)
    def decorated_function(*args, **kwargs):
        if not devenv:
            return abort(404)
        return f(*args, **kwargs)
    return decorated_function


@app.before_request
def do_before_request():
    g.users, g.users_lookup = get_users()

@app.context_processor
def inject_context():
    return dict(devenv=devenv, inventory_assets_base_url=const.INVENTORY_ASSETS_BASE_URL)

def name_check(form, field):
    if field.data != field.data.strip():
        raise ValidationError("Name must not contain leading or trailing space")
    if any(c.isdigit() for c in field.data):
        raise ValidationError("Name must not contain numbers")


# just to manually enter login_required annotation
@app.route("/login")
@login_required
def login_dummy():
    return redirect(request.args.get("redirect_url") or "/m/inventory")

@app.route("/")
def index():
    return redirect("/m/inventory")

def get_mac_metrics():
    registry = CollectorRegistry()
    inventory_item = Gauge('inventory_item_info', 'Information about inventory item', ['mac', 'slug', 'owner', 'user', 'name'], registry=registry)
    items = mongodb.inventory.find(
      filter = { "mac": { "$exists": True } },
      projection = {
        "mac": 1,
        "name": 1,
        "shortener.slug": 1,
        "inventory.owner": 1,
        "inventory.user": 1,
      }
    )
    for item in items:
        inventory_item.labels(mac=item.mac, owner=item.get("owner", {}).get("username"), user=item.get("user", {}).get("username"), slug=item.slug, name=item.name).set(1)
    return generate_latest(registry)

class MetricsHandler(BaseHTTPRequestHandler):
    def do_GET(self):
        if self.path == '/metrics':
            self.send_response(200)
            self.send_header('Content-Type', 'text/plain')
            self.end_headers()
            self.wfile.write(get_mac_metrics())
        else:
            self.send_response(404)
            self.end_headers()

class MultiCheckboxField(SelectMultipleField):
    widget = widgets.ListWidget(prefix_label=False)
    option_widget = widgets.CheckboxInput()

if __name__ == '__main__':
    if not devenv:
        metrics_address = ('', 8000)
        httpd = HTTPServer(metrics_address, MetricsHandler)
        metrics_thread = threading.Thread(target=httpd.serve_forever)
        metrics_thread.daemon = True
        metrics_thread.start()

    app.run(debug=devenv, host='::')