forked from k-space/doorboy-proxy
139 lines
4.3 KiB
Python
139 lines
4.3 KiB
Python
import os
|
|
from datetime import datetime, timezone
|
|
from typing import Tuple
|
|
|
|
import kube
|
|
import requests
|
|
from pymongo.errors import PyMongoError
|
|
from requests.exceptions import RequestException
|
|
from sanic import Blueprint
|
|
from sanic.response import text
|
|
|
|
slack_app = Blueprint("slack", __name__)
|
|
|
|
# webhook logs to private channel or "DEV" to print to console.
|
|
SLACK_DOORLOG_CALLBACK = os.environ["SLACK_DOORLOG_CALLBACK"]
|
|
# used to verify (deprecated) incoming requests from slack
|
|
SLACK_VERIFICATION_TOKEN = os.environ["SLACK_VERIFICATION_TOKEN"]
|
|
SLACK_CHANNEL_ID = os.environ["SLACK_CHANNEL_ID"] # TODO:
|
|
|
|
def fauthGroup(door: str) -> str:
|
|
match door:
|
|
case "alldoors" | "backdoor" | "frontdoor" | "grounddoor":
|
|
return "k-space:floor"
|
|
case "workshopdoor":
|
|
return "k-space:workshop"
|
|
case _:
|
|
return None
|
|
|
|
def slack_post(msg):
|
|
if SLACK_DOORLOG_CALLBACK == "DEV":
|
|
print(f"[DEV SLACK]: {msg}")
|
|
return
|
|
|
|
try:
|
|
requests.post(SLACK_DOORLOG_CALLBACK, json={"text": msg}).raise_for_status()
|
|
except RequestException as e:
|
|
print(f"[SLACK]: {e}")
|
|
|
|
|
|
def approvedStr(approved: bool) -> str:
|
|
if approved:
|
|
return "Permitted"
|
|
|
|
return "Denied"
|
|
|
|
|
|
# consumes SLACK_DOORLOG_CALLBACK and app.ctx.db
|
|
@slack_app.listener("after_server_start")
|
|
async def slack_log_fwd(app):
|
|
pipeline = [
|
|
{
|
|
"$match": {
|
|
"operationType": "insert",
|
|
}
|
|
}
|
|
]
|
|
while True:
|
|
try:
|
|
async with app.ctx.db.doorlog.watch(pipeline) as stream:
|
|
async for event in stream:
|
|
ev = event["fullDocument"]
|
|
|
|
msg = "%s %s access for %s via %s" % (
|
|
approvedStr(ev["approved"]),
|
|
ev["door"],
|
|
ev["user"],
|
|
ev["method"],
|
|
)
|
|
slack_post(msg)
|
|
|
|
except PyMongoError as e:
|
|
print(e)
|
|
|
|
# -> approved, user, err
|
|
def slack_authz(authGroup: str, slackId: str, channel_id: str) -> Tuple[bool, str, str]:
|
|
groups, user = kube.by_slackid(slackId)
|
|
if user is None:
|
|
if authGroup == "k-space:floor":
|
|
if channel_id == SLACK_CHANNEL_ID:
|
|
print(f"WARN: slack #members open with unlinked ID: {slackId}")
|
|
return True, user, f"This will stop working! Your Slack ID {slackId} is not linked with auth.k-space.ee, please notify info@k-space.ee."
|
|
|
|
return False, user, f"No user with slack_id {slackId}. Try in #members or doorboy.k-space.ee. Help at info@k-space.ee.",
|
|
else:
|
|
return False, user, f"No user with slack_id {slackId}. Try doorboy.k-space.ee. Help at info@k-space.ee."
|
|
|
|
if authGroup not in groups:
|
|
return False, user, f"You are not in {authGroup}. k-space.ee/membership"
|
|
|
|
return True, user, ""
|
|
|
|
@slack_app.route("/slack-open", methods=["POST"])
|
|
async def slack_open(request):
|
|
if request.form.get("token") != SLACK_VERIFICATION_TOKEN:
|
|
print("WARN: /slack-open route accessed with invalid token")
|
|
return "Invalid token (are you Slack?)", 401
|
|
|
|
command = request.form.get("command")
|
|
door = command.removeprefix("/open-").replace("-", "")
|
|
|
|
authGroup = fauthGroup(door)
|
|
if authGroup is None:
|
|
print(f"WARN: unknown slack door {door}")
|
|
return "Invalid door! (git.k-space.ee/k-space/doorboy-proxy)"
|
|
|
|
# user may be empty, if not linked to kube user
|
|
ok, user, err = slack_authz(
|
|
authGroup,
|
|
request.form.get("user_id"),
|
|
request.form.get("channel_id"),
|
|
)
|
|
|
|
userExtra = f"{request.form.get("user_id")} (slack u/n: {request.form.get("user_name")})" # slackName can be changed by user
|
|
|
|
doors = [door]
|
|
if door == "alldoors":
|
|
# outside non-special doors
|
|
doors = ["backdoor", "frontdoor", "grounddoor"]
|
|
|
|
for d in doors:
|
|
await request.app.ctx.db.doorlog.insert_one(
|
|
{
|
|
"method": "slack",
|
|
"timestamp": datetime.now(timezone.utc),
|
|
"door": d,
|
|
"approved": ok,
|
|
"user": user,
|
|
"userExtra": userExtra,
|
|
}
|
|
)
|
|
|
|
if not ok:
|
|
return text(err)
|
|
|
|
if err:
|
|
return text(f"Opening {door}… {err}")
|
|
|
|
return text(f"Opening {door}…")
|