All checks were successful
ci/woodpecker/manual/woodpecker Pipeline was successful
202 lines
6.1 KiB
Python
202 lines
6.1 KiB
Python
from datetime import datetime, timedelta
|
|
|
|
from bson.objectid import ObjectId
|
|
from flask import Blueprint, g, redirect, render_template, request
|
|
from flask_wtf import FlaskForm
|
|
from pymongo import MongoClient
|
|
from wtforms import StringField, IntegerField, SelectField, validators
|
|
from wtforms.validators import DataRequired
|
|
|
|
import const
|
|
from common import spam, users_lookup
|
|
from oidc import login_required, read_user
|
|
|
|
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
|
|
token = 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")
|
|
|
|
|
|
|
|
@page_doorboy.route("/m/doorboy/<token_id>/disable")
|
|
@login_required
|
|
def view_doorboy_disable(token_id):
|
|
user = read_user()
|
|
db.inventory.update_one({
|
|
"component": "doorboy",
|
|
"type": "token",
|
|
"_id": ObjectId(token_id),
|
|
"inventory.owner.username": user["username"]
|
|
}, {
|
|
"$set": {
|
|
"token.disabled": datetime.utcnow()
|
|
},
|
|
"$unset": {
|
|
"token.enabled": ""
|
|
}
|
|
|
|
})
|
|
return redirect("/m/doorboy")
|
|
|
|
@page_doorboy.route("/m/doorboy/<token_id>/enable")
|
|
@login_required
|
|
def view_doorboy_enable(token_id):
|
|
user = read_user()
|
|
db.inventory.update_one({
|
|
"component": "doorboy",
|
|
"type": "token",
|
|
"_id": ObjectId(token_id),
|
|
"inventory.owner.username": user["username"]
|
|
}, {
|
|
"$set": {
|
|
"token.enabled": datetime.utcnow(),
|
|
},
|
|
"$unset": {
|
|
"token.disabled": ""
|
|
}
|
|
})
|
|
return redirect("/m/doorboy")
|
|
|
|
|
|
class TokenEditForm(FlaskForm):
|
|
comment = StringField("Comment")
|
|
|
|
@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", "")
|
|
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,
|
|
}
|
|
})
|
|
return redirect("/m/doorboy")
|
|
|
|
class HoldDoorForm(FlaskForm):
|
|
door_name = SelectField("Door name", choices=[(j,j) for j in ["ground", "front", "back"]], validators=[DataRequired()])
|
|
duration = IntegerField('Duration in seconds', validators=[DataRequired(), validators.NumberRange(min=5, max=21600)])
|
|
|
|
@page_doorboy.route("/m/doorboy/hold", methods=["POST"])
|
|
@login_required
|
|
def view_doorboy_hold():
|
|
user = read_user()
|
|
form = HoldDoorForm(request.form)
|
|
now = datetime.utcnow()
|
|
if form.validate_on_submit():
|
|
db.eventlog.insert_one({
|
|
"component": "doorboy",
|
|
"type": "hold",
|
|
"requester": user["name"],
|
|
"door": form.door_name.data,
|
|
"expires": datetime.utcnow() + timedelta(seconds=form.duration.data)
|
|
})
|
|
return redirect("/m/doorboy")
|
|
|
|
|
|
@page_doorboy.route("/m/doorboy/<door>/open")
|
|
@login_required
|
|
def view_doorboy_open(door):
|
|
user = read_user()
|
|
if door not in ("ground", "front", "back"): raise
|
|
approved = user["username"] in users_lookup
|
|
db.eventlog.insert_one({
|
|
"method": "web",
|
|
"approved": approved,
|
|
"duration": 5,
|
|
"component": "doorboy",
|
|
"type": "open-door",
|
|
"door": door,
|
|
"member_id": user["username"],
|
|
"member": user["name"],
|
|
"timestamp": datetime.utcnow(),
|
|
})
|
|
status = "Permitted" if approved else "Denied"
|
|
subject = user["name"]
|
|
msg = "%s %s door access for %s via https://inventory.k-space.ee/m/doorboy" % (status, door, subject)
|
|
spam(msg)
|
|
return redirect("/m/doorboy")
|
|
|
|
|
|
@page_doorboy.route("/m/doorboy/slam", methods=["POST"])
|
|
@login_required
|
|
def view_doorboy_slam():
|
|
user = read_user()
|
|
db.eventlog.insert_one({
|
|
"component": "doorboy",
|
|
"type": "hold",
|
|
"requester": user["name"],
|
|
"door": form.door_name.data,
|
|
"expires": datetime.utcnow() + timedelta(minutes=form.duration_min.data)
|
|
})
|
|
return redirect("/m/doorboy")
|
|
|
|
|
|
@page_doorboy.route("/m/doorboy")
|
|
@login_required
|
|
def view_doorboy():
|
|
user = read_user()
|
|
latest_events = db.eventlog.find({"component": "doorboy", "type":"open-door"}).sort([("timestamp", -1)]).limit(10);
|
|
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/swipes")
|
|
@login_required
|
|
def view_doorboy_events():
|
|
user = read_user()
|
|
latest_events = db.eventlog.find({"component": "doorboy", "event":"card-swiped"}).sort([("timestamp", -1)]).limit(500);
|
|
return render_template("doorboy.html", **locals())
|
|
|
|
|
|
@page_doorboy.route("/m/doorboy/<token_id>/events")
|
|
@login_required
|
|
def view_doorboy_token_events(token_id):
|
|
user = read_user()
|
|
token = db.inventory.find_one({"_id": ObjectId(token_id)})
|
|
latest_events = db.eventlog.find({"component": "doorboy", "event":"card-swiped", "token.uid": token.get("token").get("uid")}).sort([("timestamp", -1)])
|
|
return render_template("doorboy.html", **locals())
|