Authorization check through OpenPolicyAgent
@@ -84,19 +84,7 @@ try:
def before_request():
g.db = dbpool.get()
- # token = get_token()
-
- input_data = json.dumps({
- "input": {
- "method": request.method,
- "path": request.path.strip().split("/")[1:-1]
- }
- })
-
- rsp = requests.post(os.environ['INFRABOX_OPA_HOST']+"/v1/data/httpapi/authz?explain=full", data=input_data)
- logger.info(input_data)
- r = rsp.content
- logger.info(r)
+ check_request_authorization()
def release_db():
db = getattr(g, 'db', None)
@@ -113,12 +101,7 @@ except:
def before_request():
g.db = DB(connect_db())
- input = jsonify({
- "method": request.method,
- "path": request.path.strip().split("/")[1:]
- # "token": require_token()
- })
- print(input)
+ check_request_authorization()
def release_db():
db = getattr(g, 'db', None)
@@ -130,6 +113,26 @@ except:
g.release_db = release_db
+def check_request_authorization():
+ try:
+ input_data = json.dumps({
+ "input": {
+ "method": request.method,
+ "path": request.path.strip().split("/")[1:-1]
+ }
+ })
+
+ rsp = requests.post(os.environ['INFRABOX_OPA_HOST']+"/v1/data/httpapi/authz", data=input_data)
+ logger.info(input_data)
+ rsp_dict = rsp.json()
+ logger.info(rsp_dict)
+
+ if not ("result" in rsp_dict and rsp_dict["result"]["allow"] is True):
+ abort(401, 'Unauthorized')
+ except requests.exceptions.RequestException:
+ abort(500, 'Authorization failed')
+
+
@app.teardown_request
def teardown_request(_):
try:
