Files
inventory-app/inventory-app/doorboy.py
2026-06-12 01:49:52 +03:00

177 lines
6.1 KiB
Python

from datetime import datetime, timedelta
import const
from bson.objectid import ObjectId
from common import User
from flask import Blueprint, abort, g, redirect, render_template, request
from flask_wtf import FlaskForm
from oidc import login_required, read_user
from pymongo import MongoClient
from wtforms import BooleanField, IntegerField, SelectField, StringField, validators
from wtforms.validators import DataRequired
page_doorboy = Blueprint("doorboy", __name__)
db = MongoClient(const.MONGO_URI).get_default_database()
@page_doorboy.route("/m/doorboy/<event_id>/claim")
@login_required
def view_doorboy_claim(event_id):
user = read_user()
# Find swipe event OR token object by id to get card UID
event = db.inventory.find_one({
"_id": ObjectId(event_id)
})
# Find token object to associate with user
db.inventory.update_one({
"type": "token",
"token.uid_hash": event["token"]["uid_hash"],
"inventory.owner.username": { "$exists": False }
}, {
"$set": {
"token.enabled": datetime.utcnow(),
"inventory.owner.display_name": user["name"],
"inventory.owner.username": user["username"]
}
})
return redirect("/m/doorboy")
class TokenEditForm(FlaskForm):
comment = StringField("Comment")
enabled = BooleanField("Enabled")
@page_doorboy.route("/m/doorboy/<token_id>/edit", methods=["GET"])
@login_required
def view_doorboy_edit(token_id):
user = read_user()
token = db.inventory.find_one({
"component": "doorboy",
"type": "token",
"_id": ObjectId(token_id),
"inventory.owner.username": user["username"]
})
form = TokenEditForm()
form.comment.data = token["token"].get("comment", "")
if token["token"].get("enabled"):
form.enabled.render_kw = {"checked": "checked"}
form.enabled.data = "y"
return render_template("doorboy_token_edit.html", form=form, token=token)
@page_doorboy.route("/m/doorboy/<token_id>/edit", methods=["POST"])
@login_required
def save_doorboy_edit(token_id):
user = read_user()
form = TokenEditForm(request.form)
if form.validate_on_submit():
db.inventory.update_one({
"component": "doorboy",
"type": "token",
"_id": ObjectId(token_id),
"inventory.owner.username": user["username"]
}, {
"$set": {
"token.comment": form.comment.data,
"token.enabled": form.enabled.data,
}
})
return redirect("/m/doorboy/me")
class HoldDoorForm(FlaskForm):
door_name = SelectField("Door name", choices=[(j,j) for j in ["grounddoor", "frontdoor", "backdoor"]], validators=[DataRequired()])
duration = IntegerField('Duration in seconds', validators=[DataRequired(), validators.NumberRange(min=5, max=21600)])
# duration=0 to override and close right away
@page_doorboy.route("/m/doorboy/hold", methods=["POST"])
@login_required
def view_doorboy_hold():
user = read_user()
form = HoldDoorForm(request.form)
if form.validate_on_submit():
db.doorlog.insert_one({
"method": "hold",
"timestamp": datetime.utcnow(),
"door": form.door_name.data,
"approved": True,
"user": user["username"],
"expires": datetime.utcnow() + timedelta(seconds=form.duration.data)
})
return redirect("/m/doorboy")
# Writes open event to log, which is picked up by doorboy-proxy.
@page_doorboy.route("/m/doorboy/<door>/open")
@login_required
def view_doorboy_open(door):
user = read_user()
if door not in ("grounddoor", "frontdoor", "backdoor", "workshopdoor"):
return "", 400
if door == "workshopdoor":
access_group = "k-space:workshop"
else:
access_group = "k-space:floor"
approved = access_group in g.users_lookup.get(user["username"], User()).groups
db.doorlog.insert_one({
"method": "web",
"timestamp": datetime.utcnow(),
"door": door,
"approved": approved,
"user": user["username"],
})
if approved:
return redirect("/m/doorboy")
else:
return "", 401
@page_doorboy.route("/m/doorboy")
@login_required
def view_doorboy():
user = read_user()
workshop_access = "k-space:workshop" in g.users_lookup.get(user["username"], User()).groups
latest_swipes = db.inventory.find({"component": "doorboy", "type":"token"}).sort([("last_seen", -1)]).limit(10)
return render_template("doorboy.html", **locals())
@page_doorboy.route("/m/doorboy/me")
@login_required
def view_own_cards():
user = read_user()
return view_user_cards_inner(user["username"])
def view_user_cards_inner(username):
user = read_user()
subject_user = g.users_lookup.get(username)
if not subject_user:
return abort(404)
is_self = user["username"] == subject_user["username"]
cards = db.inventory.find({
"component": "doorboy",
"type":"token",
"inventory.owner.username": username
}).sort([("last_seen", -1)])
return render_template("doorboy_user.html", **locals())
@page_doorboy.route("/m/doorboy/log")
@login_required(groups=["k-space:board", "k-space:kubernetes:admins"])
def doorlog():
latest_events = db.doorlog.find({}).sort([("timestamp", -1)])
return render_template("doorboy_log.html", latest_events=latest_events)
@page_doorboy.route("/m/doorboy/log/<username>")
@login_required(groups=["k-space:board", "k-space:kubernetes:admins"])
def doorlog_user(username):
if username is None:
return redirect("/m/doorboy/log")
latest_events = db.doorlog.find({"user": username}).sort([("timestamp", -1)])
return render_template("doorboy_log.html", latest_events=latest_events, info="Slack users without linked IDs opening from #members appear in <a href=\"/m/doorboy/log/_\">Unlinked users</a>.")
@page_doorboy.route("/m/doorboy/log/_")
@login_required(groups=["k-space:board", "k-space:kubernetes:admins"])
def doorlog_unknowns():
latest_events = db.doorlog.find({"user": {"$in": [None, ""]}}).sort([("timestamp", -1)])
return render_template("doorboy_log.html", latest_events=latest_events, info="Showing opens not linked to kube users.")