"""Logic for handling per-endpoint rules.""" import re import yaml class EndpointRuleType(object): r"""Endpoint rule type constants. FLASK ----- Provides flask-like format for endpoint rules. You can use the following dynamic route args that will be converted to the corresponding regex operators: - match any numeric value - match any string Examples: '/holds///average' Please, not use arg names like you'd do in Flask ()! REGEX ----- Plain regex rule type. Example: '^\\/holds\\/\\w+\\/\\d+$' """ FLASK = 'flask' REGEX = 'regex' class EndpointRule(object): """Class that encapsulates an endpoint rule.""" def __init__( self, path: str, methods: list = ['*'], roles: dict = {}, rule_type: str = EndpointRuleType.FLASK): """Create a new EndpointRule. Args: path (str): regex expression, defining the path methods (list): a list of allowed HTTP methods roles (dict): profile type as a key, roles list as a value """ self.path = path if rule_type == EndpointRuleType.FLASK: path = route_to_regex(path) self.path_regex = re.compile(path) self.methods = set(methods) self.roles = roles self.rule_type = rule_type def match(self, path: str) -> bool: """Check if provided path matches this rules path. Args: path (str): endpoint path to check Returns: bool: matched or not """ return self.path_regex.match(path) def is_allowed(self, method: str, profile_type: str, roles: [int]) -> bool: """Check access for specified method and role. Args: method (str): HTTP method to check profile_type (str): can be 'oa' or 'alw' roles (list[int]): roles to check Returns: bool: allows access or not """ if '*' in self.methods or method in self.methods: allowed_roles = set(self.roles.get(profile_type, [])) return ( '*' in allowed_roles or allowed_roles.intersection(set(roles))) else: return False def from_yaml(rule: dict) -> 'EndpointRule': """Create EndpointRule instance from parsed yaml. Args: rule (dict): a rule dictionary from yaml file Returns: EndpointRule: a new instance of the rule class. """ return EndpointRule( path=rule['path'], methods=rule['methods'], roles=rule['profiles'], rule_type=rule.get('rule_type', EndpointRuleType.FLASK)) def route_to_regex(route: str) -> str: """Convert Flask-like route to a valid regex pattern. Args: route (str): route to convert Returns: str: regex pattern """ remap = { '': '\\d+', '': '[\\w_\\-]+', } for token, rx in remap.items(): route = route.replace(token, rx) return f'^{route}$' class EndpointRulesValidator(object): """Validator for checking role, path and method with rules list.""" def __init__(self, yaml_file: str): """Constructor.""" self.rules = self._load_rules_yaml(yaml_file) def _load_rules_yaml(self, yaml_file: str) -> [EndpointRule]: """Load rules list from yaml file. Args: yaml_file (str): yaml file name Returns: [EndpointRule]: a list of EndpointRule objects """ with open(yaml_file, 'r') as f: yaml_rules = yaml.load(f, Loader=yaml.FullLoader) rules = yaml_rules['rules'] return [from_yaml(r) for r in rules] def _find_rules(self, path: str) -> [EndpointRule]: """Try to find a list of EndpointRules by endpoint path. Args: path (str): endpoint path to search for Returns: list[EndpointRule]: an list of endpoint rules """ path = path.rstrip('/') rules = [r for r in self.rules if r.match(path)] return rules def has_access( self, path: str, method: str, profile: str, roles: [int]) -> bool: """Check access for the specified path, method and role. Args: path (str): endpoint path method (str): HTTP method profile (str): profile type (can be 'alw' or 'oa') roles (list[int]): user roles to check Returns: bool: has access or not """ rules = self._find_rules(path) if rules: return any([r.is_allowed(method, profile, roles) for r in rules]) else: return False