You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
120 lines
3.4 KiB
120 lines
3.4 KiB
import json
|
|
from functools import wraps
|
|
|
|
from django.db.models import Q
|
|
from django.http import HttpResponse, JsonResponse
|
|
from hc.accounts.models import Project
|
|
from hc.lib.jsonschema import ValidationError, validate
|
|
|
|
|
|
def error(msg, status=400):
|
|
return JsonResponse({"error": msg}, status=status)
|
|
|
|
|
|
def authorize(f):
|
|
@wraps(f)
|
|
def wrapper(request, *args, **kwds):
|
|
if "HTTP_X_API_KEY" in request.META:
|
|
api_key = request.META["HTTP_X_API_KEY"]
|
|
elif hasattr(request, "json"):
|
|
api_key = str(request.json.get("api_key", ""))
|
|
else:
|
|
api_key = ""
|
|
|
|
if len(api_key) != 32:
|
|
return error("missing api key", 401)
|
|
|
|
try:
|
|
request.project = Project.objects.get(api_key=api_key)
|
|
except Project.DoesNotExist:
|
|
return error("wrong api key", 401)
|
|
|
|
request.readonly = False
|
|
return f(request, *args, **kwds)
|
|
|
|
return wrapper
|
|
|
|
|
|
def authorize_read(f):
|
|
@wraps(f)
|
|
def wrapper(request, *args, **kwds):
|
|
if "HTTP_X_API_KEY" in request.META:
|
|
api_key = request.META["HTTP_X_API_KEY"]
|
|
elif hasattr(request, "json"):
|
|
api_key = str(request.json.get("api_key", ""))
|
|
else:
|
|
api_key = ""
|
|
|
|
if len(api_key) != 32:
|
|
return error("missing api key", 401)
|
|
|
|
write_key_match = Q(api_key=api_key)
|
|
read_key_match = Q(api_key_readonly=api_key)
|
|
try:
|
|
request.project = Project.objects.get(write_key_match | read_key_match)
|
|
except Project.DoesNotExist:
|
|
return error("wrong api key", 401)
|
|
|
|
request.readonly = api_key == request.project.api_key_readonly
|
|
return f(request, *args, **kwds)
|
|
|
|
return wrapper
|
|
|
|
|
|
def validate_json(schema={"type": "object"}):
|
|
""" Parse request json and validate it against `schema`.
|
|
|
|
Put the parsed result in `request.json`.
|
|
If schema is None then only parse and check if the root
|
|
element is a dict. Supports a limited subset of JSON schema spec.
|
|
|
|
"""
|
|
|
|
def decorator(f):
|
|
@wraps(f)
|
|
def wrapper(request, *args, **kwds):
|
|
if request.method == "POST" and request.body:
|
|
try:
|
|
request.json = json.loads(request.body.decode())
|
|
except ValueError:
|
|
return error("could not parse request body")
|
|
else:
|
|
request.json = {}
|
|
|
|
try:
|
|
validate(request.json, schema)
|
|
except ValidationError as e:
|
|
return error("json validation error: %s" % e)
|
|
|
|
return f(request, *args, **kwds)
|
|
|
|
return wrapper
|
|
|
|
return decorator
|
|
|
|
|
|
def cors(*methods):
|
|
methods = set(methods)
|
|
methods.add("OPTIONS")
|
|
methods_str = ", ".join(methods)
|
|
|
|
def decorator(f):
|
|
@wraps(f)
|
|
def wrapper(request, *args, **kwds):
|
|
if request.method == "OPTIONS":
|
|
# Handle OPTIONS here
|
|
response = HttpResponse(status=204)
|
|
elif request.method in methods:
|
|
response = f(request, *args, **kwds)
|
|
else:
|
|
response = HttpResponse(status=405)
|
|
|
|
response["Access-Control-Allow-Origin"] = "*"
|
|
response["Access-Control-Allow-Headers"] = "X-Api-Key"
|
|
response["Access-Control-Allow-Methods"] = methods_str
|
|
response["Access-Control-Max-Age"] = "600"
|
|
return response
|
|
|
|
return wrapper
|
|
|
|
return decorator
|