Authorization core module

kingdom/access/authentication/authenticator.py

@@ -0,0 +1,50 @@
+"""
+authenticator.py
+Handles authentication throughout services
+"""
+from dataclasses import dataclass
+from typing import Dict
+
+import jwt
+
+from src.core import config
+
+
+class InvalidToken(Exception):
+    def __init__(self):
+        super().__init__("Invalid JWT token.")
+
+
+@dataclass
+class JWT:
+    "Simple JWT Wrapper"
+    token: str
+
+    def decode(self) -> Dict:
+        """Decodes a JWT"""
+        try:
+            self.payload = jwt.decode(
+                jwt=self.token,
+                key=config.get_jwt_secret_key(),
+                algorithms=config.JWT_ALGORITHM,
+            )
+            return self.payload
+        except jwt.PyJWTError:
+            raise InvalidToken()
+
+
+class Authenticator:
+    token: JWT
+    payload: Dict
+
+    def __init__(self, token: str):
+        self.jwt = JWT(token)
+
+    def __call__(self):
+        self.payload = self.jwt.decode()
+
+
+def test_authenticator():
+    token = "abr9f8c09adj!lsd.sdxxf.ldfj"
+    auth = Authenticator(token)
+    assert auth()

kingdom/access/authentication/types.py

@@ -0,0 +1,82 @@
+from datetime import datetime, timedelta
+from typing import Dict, Generator, List, Optional, Tuple
+
+import jwt
+
+from kingdom.access.authorization.encode import encode
+from kingdom.access.authorization.types import (
+    AccessRequest,
+    Conditional,
+    Permission,
+    Policy,
+    PolicyContext,
+    Resource,
+    dataclass,
+)
+
+RANDOM_KEY = "abcd00f"
+JWT_ALGORITHM = "HS256"
+
+TOKEN_EXPIRATION_MIN = 30
+
+
+@dataclass
+class Role:
+    name: str
+    policies: List[Optional[Policy]]
+
+    def __hash__(self) -> int:
+        return hash(self.name)
+
+
+@dataclass
+class User:
+    access_key: str
+    roles: List[Optional[Role]]
+
+    def resolve_policies(self):
+        # TODO: Improve this typing
+        for role in self.roles:
+            for policy in role.policies:
+                yield policy
+
+
+JWTDecodedPayload = Dict
+
+MaybeBytes = Tuple[Optional[bytes], Optional[Exception]]
+MaybePayload = Tuple[Optional[JWTDecodedPayload], Optional[Exception]]
+
+
+def encode_user_policies(user: User) -> PolicyContext:
+    user_policies = list(user.resolve_policies())
+    return encode(user_policies)
+
+
+def encode_user_payload(user: User) -> JWTDecodedPayload:
+    expiration = datetime.utcnow() + timedelta(minutes=TOKEN_EXPIRATION_MIN)
+    return dict(
+        sub=user.access_key, exp=expiration, roles=encode_user_policies(user),
+    )
+
+
+def encode_jwt(user: User) -> MaybeBytes:
+    payload = encode_user_payload(user)
+    try:
+        return (
+            jwt.encode(
+                payload=payload, key=RANDOM_KEY, algorithm=JWT_ALGORITHM
+            ),
+            None,
+        )
+    except jwt.PyJWTError as exc:
+        return None, exc
+
+
+def decode_jwt(token: bytes) -> MaybePayload:
+    try:
+        return (
+            jwt.decode(jwt=token, key=RANDOM_KEY, algorithms=[JWT_ALGORITHM]),
+            None,
+        )
+    except jwt.PyJWTError as exc:
+        return None, exc

kingdom/access/authorization/dsl.py

@@ -0,0 +1,201 @@
+# test_authorization.py
+""""
+dsl.py
+
+ SPECIFICATION of Conditionals:
+
+ Vocabulary. These contains current implementation expected limitations and
+ simplifications:
+   primary     ::= "resource"
+   identifier  ::= "id"
+   selector    ::= "*" | string
+   attrref     ::= primary "." identifier
+   or_expr     ::= "||"
+   compr_op    ::= "=="
+   cond        ::= attrref compr_op selector
+   conds       ::= (cond or_expr)*
+
+ There are two available selectors:
+   1. All instances:        "*"
+   2. Individual instances: "<uuid>"
+
+ There are some conditions to selectors:
+   1. All selector must **always** be alone.
+
+OBS.: This is in WIP and should be thoroughly simplified & documented.
+"""
+
+import string
+from typing import List
+
+
+def conditionals_split(sequence: str):
+    expressions = sequence.split("||")
+    if "" in expressions:
+        # meaning that we had a loose OR
+        return False
+
+    parsed_expr = [expression.strip() for expression in expressions]
+    for expr in parsed_expr:
+        for char in expr:
+            if char.isspace():
+                # illegal whitespaces
+                return False
+
+            if char in set(string.punctuation):
+                # illegal characteres
+                return False
+
+    return tuple(parsed_expr)
+
+
+def parse_identifier(expression):
+    # since it's the beginning, we can strip this, to avoid overly
+    # complicated iterations :-)
+    expression = expression.strip()
+    identifier = ""
+    for idx, token in enumerate(expression):
+        if token.isidentifier():
+            identifier += token
+        else:
+            if token == "." and len(identifier) > 0:
+                return identifier, expression[idx:]
+            return (False,)
+
+
+VALID_OPS = {"==", ">", "<", ">=", "<=", "!="}
+VALID_OPS_TOKEN = {token for operator in VALID_OPS for token in operator}
+
+
+def parse_reference(reference_expr):
+    reference = ""
+    parsing_idx = -1
+
+    # first we try to read all valid tokens after "." and before a valid
+    # operator
+    for idx, token in enumerate(reference_expr):
+        if idx == 0:
+            # first token **must** be a ".", no other one allowed
+            if token != ".":
+                return (False,)
+            parsing_idx = idx
+            continue
+
+        if token.isspace() or token in VALID_OPS_TOKEN:
+            # we have found our potential reference
+            parsing_idx = idx
+            break
+
+        if token.isidentifier():
+            reference += token
+        else:
+            return (False,)
+
+    # check for illegality on the rest of the expression
+    # meaning we must only accept white spaces between end of ref and operator
+    rest = reference_expr[parsing_idx:]
+    for idx, token in enumerate(rest):
+        if token in VALID_OPS_TOKEN:
+            return (reference, rest[idx:])
+
+        if token.isspace():
+            # the only acceptable token between a ref and operator
+            continue
+        else:
+            return (False,)
+
+
+def parse_identifier_reference(expression):
+    identifier, *reference_expr = parse_identifier(expression)
+    if identifier is False:
+        return False
+    reference, *operator_expr = parse_reference(*reference_expr)
+    if reference is False:
+        return False
+
+    return (identifier.upper(), reference.upper())
+
+
+def parse_operator(operator_expr):
+    operator_expr = operator_expr.strip()  # just to make sure
+
+    operator = ""
+    parsing_idx = -1
+    # First we parse a known operator.
+    for idx, token in enumerate(operator_expr):
+        if token in VALID_OPS_TOKEN:
+            # the only valid thing that is being parsed
+            operator += token
+        else:
+            parsing_idx = idx
+            break
+
+    # Then we check if the rest is OK.
+    rest = operator_expr[parsing_idx:]
+    for idx, token in enumerate(rest):
+        if token == "'":
+            # valid stopping point
+            if operator in VALID_OPS:
+                return (operator, rest[idx:])
+            # invalid operator
+            return (False,)
+
+        if token.isspace():
+            continue
+        else:
+            # illegal characteres
+            return (False,)
+
+
+def parse_selector(selector_expr):
+    ALL_TOKEN = "*"
+    selector_expr = selector_expr.strip()
+    selector = ""
+    parsing_idx = -1
+
+    def isselector(token):
+        return token.isnumeric() or token.isidentifier() or token == ALL_TOKEN
+
+    # First we try to find a selector.
+    for idx, token in enumerate(selector_expr):
+        if idx == 0:
+            # first token
+            if token != "'":
+                return (False,)
+            continue
+
+        if token == "'":
+            parsing_idx = idx + 1
+            break
+
+        if isselector(token):
+            selector += token
+        else:
+            return (False,)
+
+    # Edge case: are we dealig with an *?
+    if ALL_TOKEN in selector and selector != "*":
+        # plain comparison
+        return (False,)
+
+    rest = selector_expr[parsing_idx + 1:]
+    if len(rest) > 0 or len(selector) == 0:
+        # we shouldn't have anything left
+        return (False,)
+    return (selector,)
+
+
+def parse_expression(expr):
+    identifier, *reference_expr = parse_identifier(expr)
+    if identifier is False:
+        return False
+    reference, *operator_expr = parse_reference(*reference_expr)
+    if reference is False:
+        return False
+    operator, *selector_expr = parse_operator(*operator_expr)
+    if operator is False:
+        return False
+    selector, *end = parse_selector(*selector_expr)
+    if selector is False:
+        return False
+    return (identifier, reference, operator, selector)