932b0b48-8d1a-4bbd-a9fd-a090d6114750

Build # 932b0b48-8d1a-4bbd-a9fd-a090d6114750

Build Type

push

circleci

Committed by Sym Roe

Commit Message

Use S3 ballot cache rather than WCIVF directly

Run Details

507 of 582 relevant lines covered (87.11%)

0.87 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

78.26

/api/api_auth/handler.py

from common.auth_models import User, UserDoesNotExist
from common.sentry_helper import init_sentry

init_sentry()


def dynamodb_auth(api_key: str, region_name="eu-west-2"):
    ret = {
        "authenticated": False,
        "error": None,
        "warnings": [],
        "data": {},
    }
    try:
        user = User.from_dynamodb(api_key)
    except UserDoesNotExist:
        ret["error"] = "API key not found"
        return ret
    ret.update({"data": user.as_dict()})
    if user.is_active:
        ret["authenticated"] = True
    else:
        ret["error"] = "API key not active"
    if user.rate_limit_warn:
        ret["warnings"].append("Rate limit exceeded")

    return ret


def lambda_handler(event, context):
    if "auth_token" not in event["queryStringParameters"]:
        raise Exception("Unauthorized")
    api_key = event["queryStringParameters"].get("auth_token", None)
    authentication = {"data": {"auth_token": api_key}}
    # TMP: disable dynamodb lookup for the time being
    # if not api_key:
    #     raise Exception("Unauthorized")
    #
    # authentication = dynamodb_auth(api_key)
    # if not authentication["authenticated"]:
    #     raise Exception("Unauthorized")

    return {
        "principalId": "sym",
        "policyDocument": {
            "Version": "2012-10-17",
            "Statement": [
                {
                    "Action": "execute-api:Invoke",
                    "Effect": "Allow",
                    "Resource": "*",
                }
            ],
        },
        "context": authentication["data"],
    }

1	from common.auth_models import User, UserDoesNotExist	1✔
2	from common.sentry_helper import init_sentry	1✔
3
4	init_sentry()	1✔
5
6
7	def dynamodb_auth(api_key: str, region_name="eu-west-2"):	1✔
8	ret = {	1✔
9	"authenticated": False,
10	"error": None,
11	"warnings": [],
12	"data": {},
13	}
14	try:	1✔
15	user = User.from_dynamodb(api_key)	1✔
16	except UserDoesNotExist:	1✔
17	ret["error"] = "API key not found"	1✔
18	return ret	1✔
19	ret.update({"data": user.as_dict()})	1✔
20	if user.is_active:	1✔
21	ret["authenticated"] = True	1✔
22	else:
23	ret["error"] = "API key not active"	1✔
24	if user.rate_limit_warn:	1✔
25	ret["warnings"].append("Rate limit exceeded")	1✔
26
27	return ret	1✔
28
29
30	def lambda_handler(event, context):	1✔
31	if "auth_token" not in event["queryStringParameters"]:	×
32	raise Exception("Unauthorized")	×
33	api_key = event["queryStringParameters"].get("auth_token", None)	×
34	authentication = {"data": {"auth_token": api_key}}	×
35	# TMP: disable dynamodb lookup for the time being
36	# if not api_key:
37	# raise Exception("Unauthorized")
38	#
39	# authentication = dynamodb_auth(api_key)
40	# if not authentication["authenticated"]:
41	# raise Exception("Unauthorized")
42
43	return {	×
44	"principalId": "sym",
45	"policyDocument": {
46	"Version": "2012-10-17",
47	"Statement": [
48	{
49	"Action": "execute-api:Invoke",
50	"Effect": "Allow",
51	"Resource": "*",
52	}
53	],
54	},
55	"context": authentication["data"],
56	}

DemocracyClub / aggregator-api / 932b0b48-8d1a-4bbd-a9fd-a090d6114750

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous