88 lines
2.7 KiB
Python
88 lines
2.7 KiB
Python
# Copyright (c) 2020, JINGROW
|
|
# For license information, please see license.txt
|
|
|
|
from contextlib import suppress
|
|
|
|
import jingrow
|
|
|
|
|
|
def whitelist_saas_api(func): # noqa: C901
|
|
def whitelist_wrapper(fn):
|
|
return jingrow.whitelist(allow_guest=True, methods=["POST"])(fn)
|
|
|
|
def auth_wrapper(*args, **kwargs):
|
|
headers = jingrow.request.headers
|
|
site_access_token = headers.get("x-site-access-token")
|
|
site_user = headers.get("x-site-user")
|
|
site = None
|
|
site_token = None
|
|
# check when x-site-access-token is provided
|
|
if site_access_token:
|
|
splitted = site_access_token.split(":")
|
|
if len(splitted) != 2:
|
|
jingrow.throw("Invalid x-site-access-token provided", jingrow.AuthenticationError)
|
|
accessTokenDocName = splitted[0]
|
|
token = splitted[1]
|
|
with suppress(jingrow.DoesNotExistError):
|
|
record = jingrow.get_pg("Site Access Token", accessTokenDocName)
|
|
if record.token != token:
|
|
jingrow.throw("Invalid x-site-access-token provided", jingrow.AuthenticationError)
|
|
# set site and site token from access token record
|
|
site = record.site
|
|
site_token = jingrow.db.get_value("Site", site, "saas_communication_secret")
|
|
# check when x-site and x-site-token are provided
|
|
else:
|
|
# set site and site token from headers
|
|
site = headers.get("x-site")
|
|
site_token = headers.get("x-site-token")
|
|
|
|
# check for valid values
|
|
if not site or not site_token:
|
|
jingrow.throw(
|
|
"(x-site and x-site-token) or x-site-access-token headers are mandatory",
|
|
jingrow.AuthenticationError,
|
|
)
|
|
|
|
# validate site
|
|
site_record = jingrow.get_value(
|
|
"Site",
|
|
site,
|
|
[
|
|
"name",
|
|
"team",
|
|
"is_standby",
|
|
"standby_for_product",
|
|
"saas_communication_secret",
|
|
],
|
|
as_dict=True,
|
|
ignore=True,
|
|
)
|
|
|
|
if not site_record:
|
|
jingrow.throw("Invalid x-site provided", jingrow.AuthenticationError)
|
|
|
|
if site_record.saas_communication_secret != site_token:
|
|
jingrow.throw("Invalid x-site-token provided", jingrow.AuthenticationError)
|
|
|
|
if site_record.is_standby is None and site_record.standby_for_product is None:
|
|
jingrow.throw("Sorry, this is not a SaaS site", jingrow.AuthenticationError)
|
|
|
|
# set site and team name in context
|
|
jingrow.local.site_name = site_record.name
|
|
jingrow.local.team_name = site_record.team
|
|
|
|
# set team user as current user
|
|
jingrow.set_user(site_user)
|
|
|
|
# set utility function to get team and site info
|
|
jingrow.local.get_site = lambda: jingrow.get_pg("Site", jingrow.local.site_name)
|
|
jingrow.local.get_team = lambda: jingrow.get_pg("Team", jingrow.local.team_name)
|
|
|
|
# remove cmd from kwargs
|
|
kwargsCopy = kwargs.copy()
|
|
kwargsCopy.pop("cmd", None)
|
|
|
|
return func(*args, **kwargsCopy)
|
|
|
|
return whitelist_wrapper(auth_wrapper)
|