<commit_before>from .vars import *
import base64
import urlparse

class BaseAccessToken:
    """An access token (or set of tokens) that can be carried by a request"""

    def __init__(self):
        self.identifier = None

    def __repr__(self):
        return "%s(0x%x)" % (self.__class__.__name__, id(self))
        
    def _getIdentifiers(self):
        """Internal method - Returns a list of all token IDs of this token (and any subtokens it contains)"""
        return [self.identifier]
        
    def _hasExternalRepresentationFor(self, url):
        """Internal method - return True if this token can be represented externally and _getExternalRepresentation can be called"""
        return False
        
    def _getExternalRepresentation(self):
        """Internal method - return the external representation of this token"""

    def _getExternalContent(self):
        """Internal method - return key/value pairs that are important for external representation"""
        return {}
        
    def _allows(self, operation, accessChecker):
        """Internal method - return True if this token allows 'operation' on the element represented by 'accessChecker'"""
        if DEBUG: print 'access: %s %s: no access at all allowed by %s' % (operation, accessChecker.destination, self)
        return False
        
    def _allowsDelegation(self, path, rights, aud=None):
        """Internal method - return True if the given path/rights are a subset of this token, and if this token can be delegated"""
        return False
        
    def _getTokenWithIdentifier(self, identifier):
        """Internal method - return the individual (sub)token with the given ID or None"""
        return None
        
    def _getTokenDescription(self):
        """Returns a list with descriptions of all tokens in this tokenset"""
        return [dict(cid=self.identifier)]
        
    def _addChild(self, childId):
        """Register a new child token to this one"""
        assert 0

    def _delChild(self, childId):
        """Unregister a child token"""
        
    def _save(self):
        """Saves a token back to stable storage"""
        assert 0
        
    def _revoke(self):
        """Revoke this token"""
        assert 0
        
    def _getOwner(self):
        """Is the current carrier the owner of this token?"""
        return False
        
    def _setOwner(self, newOwner):
        """Set new owner of this token"""
        assert 0

    def _getObject(self):
        """Returns the object to which this token pertains"""
        return None
        
    def _removeToken(self, tokenId):
        """Remove token tokenId from this set"""
        assert 0
        
    def addToHeadersFor(self, headers, url):
        """Add this token to the (http request) headers if it has an external representation for this destination"""
        pass

    def addToHeadersAsOTP(self, headers):
        """Add this token to the (http request) headers in one-time-password form, for internal Igor use only"""
        otp = singleton.produceOTPForToken(self)
        headers['Authorization'] = 'Basic ' + base64.b64encode(otp)
        
class IgorAccessToken(BaseAccessToken):
    """A token without an external representation that allows everything everywhere.
    To be used sparingly by Igor itself."""

    def __init__(self):
        self.identifier = '*SUPER*'
        
    def _allows(self, operation, accessChecker):
        if not operation in NORMAL_OPERATIONS:
            if DEBUG: print 'access: %s %s: not allowed by supertoken' % (operation, accessChecker.destination)
        if DEBUG: print 'access: %s %s: allowed by supertoken' % (operation, accessChecker.destination)
        return True

    def _allowsDelegation(self, path, rights, aud=None):
        """Internal method - return True, the supertoken is the root of all tokens"""
        return False

class AccessToken(BaseAccessToken):
    """An access token (or set of tokens) that can be carried by a request"""

    def __init__(self, content, defaultIdentifier=None, owner=None):
        BaseAccessToken.__init__(self)
        self.owner = owner
        self.content = dict(content)
        toDelete = []
        for k in self.content:
            if not self.content[k]:
                toDelete.append(k)
        for k in toDelete:
            del self.content[k]
        #
        # Determine identifier
        #
        if defaultIdentifier == None:
            defaultIdentifier = 'no-id-%x' % id(self)
        self.identifier = content.get('cid', defaultIdentifier)
        singleton._registerTokenWithIdentifier(self.identifier, self)
        #
        # Check whether this capability is meant for this igor (no aud or aud matches our URL)
        #
        if 'aud' in content:
            audience = content['aud']
            ourUrl = singleton._getSelfAudience()
            self.validForSelf = (audience == ourUrl)
            if DEBUG: print 'access: <aud> matches: %s' % self.validForSelf
        else:
            self.validForSelf = True
        if DEBUG:  print 'access: Created:', repr(self)
        
    def __repr__(self):
        return "%s(0x%x, %s)" % (self.__class__.__name__, id(self), repr(self.content))
        
    def _hasExternalRepresentationFor(self, url):
        if not 'iss' in self.content:
            return False
        if not 'aud' in self.content:
            return False
        if url.startswith(self.content['aud']):
            return True
        p = urlparse.urlparse(url)
        if p.netloc == self.content['aud']:
            return True
        return False

    def _getExternalContent(self):
        rv = dict(cid=self.identifier)
        for key in ['obj', 'iss', 'aud', 'sub', 'nvb', 'nva', 'get', 'put', 'post', 'delete']:
            if key in self.content:
                rv[key] = self.content[key]
        return rv
        
    def _getExternalRepresentation(self):
        tokenContent = self._getExternalContent()
        externalRepresentation = singleton._encodeOutgoingData(tokenContent)
        return externalRepresentation
        
    def _allows(self, operation, accessChecker):
        # First check this this capability is for us.
        if not self.validForSelf:
            if DEBUG: print 'access: Not for this Igor: AccessToken %s' % self
            return False
        cascadingRule = self.content.get(operation)
        if not cascadingRule:
            if DEBUG: print 'access: %s %s: no %s access allowed by AccessToken %s' % (operation, accessChecker.destination, operation, self)
            return False
        path = self.content.get('obj')
        if not path:
            if DEBUG: print 'access: %s %s: no path-based access allowed by AccessToken %s' % (operation, accessChecker.destination, self)
            return False
        dest = accessChecker.destination
        destHead = dest[:len(path)]
        destTail = dest[len(path):]
        if cascadingRule == 'self':
            if dest != path:
                if DEBUG: print 'access: %s %s: does not match cascade=%s for path=%s' % (operation, dest, cascadingRule, path)
                return False
        elif cascadingRule == 'descendant-or-self':
            if destHead != path or destTail[:1] not in ('', '/'):
                if DEBUG: print 'access: %s %s: does not match cascade=%s for path=%s' % (operation, dest, cascadingRule, path)
                return False
        elif cascadingRule == 'descendant':
            if destHead != path or destTail[:1] != '/':
                if DEBUG: print 'access: %s %s: does not match cascade=%s for path=%s' % (operation, dest, cascadingRule, path)
                return False
        elif cascadingRule == 'child':
            if destHead != path or destTail[:1] != '/' or destTail.count('/') != 1:
                if DEBUG: print 'access: %s %s: does not match cascade=%s for path=%s' % (operation, dest, cascadingRule, path)
                return False
        else:
            raise AccessControlError('Capability has unknown cascading rule %s for operation %s' % (cascadingRule, operation))
        if DEBUG: print 'access: %s %s: allowed by AccessToken %s' % (operation, accessChecker.destination, self)
        return True

    def _allowsDelegation(self, newPath, newRights, aud=None):
        """Internal method - return True if the given path/rights are a subset of this token, and if this token can be delegated"""
        # Check whether this token can be delegated
        canDelegate = self.content.get('delegate')
        if not canDelegate:
            if DEBUG_DELEGATION: print 'access: delegate %s: no delegation right on AccessToken %s' % (newPath, self)
            return False
        # Check whether this is the external-supertoken and whether we want to create a new external token
        print 'xxxjack allowsDelegation', self, 'canDelegate', canDelegate, 'aud', aud
        if canDelegate == 'external':
            if aud and aud != singleton._getSelfAudience():
                # xxxjack no further checks for external tokens. This may need refining.
                return True
            else:
                return False
        else:
            if aud and aud != singleton._getSelfAudience():
                return False
        # Check whether the path is contained in our path
        path = self.content.get('obj')
        if not path:
            if DEBUG_DELEGATION: print 'access: delegate %s: no path-based access allowed by AccessToken %s' % (newPath, self)
            return False
        subPath = newPath[len(path):]
        if not newPath.startswith(path) or not subPath[:1] in ('', '/'):
            if DEBUG_DELEGATION: print 'access: delegate %s: path not contained within path for AccessToken %s' % (newPath, self)
            return False
        newIsSelf = subPath == ''
        newIsChild = subPath.count('/') == 1
        # Check that the requested rights match
        for operation, newCascadingRule in newRights.items():
            if not newCascadingRule:
                # If we don't want this operation it is always okay.
                continue
            oldCascadingRule = self.content.get(operation)
            if not oldCascadingRule:
                # If the operation isn't allowed at all it's definitely not okay.
                if DEBUG_DELEGATION: print 'access: delegate %s: no %s access allowed by AccessToken %s' % (newPath, operation, self)
                return False
            if newIsSelf:
                if not newCascadingRule in CASCADING_RULES_IMPLIED.get(oldCascadingRule, {}):
                    if DEBUG_DELEGATION: print 'access: delegate %s: %s=%s not allowed by %s=%s for AccessToken %s' % (newPath, operation, newCascadingRule, operation, oldCascadingRule, self)
                    return False
            elif newIsChild:
                # xxxjack for now only allow if original rule includes all descendants
                if not oldCascadingRule in ('descendant', 'descendant-or-self'):
                    if DEBUG_DELEGATION: print 'access: delegate %s: %s=%s not allowd by %s=%s for AccessToken %s (xxxjack temp)' % (newPath, operation, newCascadingRule, operation, oldCascadingRule, self)<commit_msg>more places where _getSelfAudience needs to be replaced with getSelfAudience<commit_after>from .vars import *
import base64
import urlparse

class BaseAccessToken:
    """An access token (or set of tokens) that can be carried by a request"""

    def __init__(self):
        self.identifier = None

    def __repr__(self):
        return "%s(0x%x)" % (self.__class__.__name__, id(self))
        
    def _getIdentifiers(self):
        """Internal method - Returns a list of all token IDs of this token (and any subtokens it contains)"""
        return [self.identifier]
        
    def _hasExternalRepresentationFor(self, url):
        """Internal method - return True if this token can be represented externally and _getExternalRepresentation can be called"""
        return False
        
    def _getExternalRepresentation(self):
        """Internal method - return the external representation of this token"""

    def _getExternalContent(self):
        """Internal method - return key/value pairs that are important for external representation"""
        return {}
        
    def _allows(self, operation, accessChecker):
        """Internal method - return True if this token allows 'operation' on the element represented by 'accessChecker'"""
        if DEBUG: print 'access: %s %s: no access at all allowed by %s' % (operation, accessChecker.destination, self)
        return False
        
    def _allowsDelegation(self, path, rights, aud=None):
        """Internal method - return True if the given path/rights are a subset of this token, and if this token can be delegated"""
        return False
        
    def _getTokenWithIdentifier(self, identifier):
        """Internal method - return the individual (sub)token with the given ID or None"""
        return None
        
    def _getTokenDescription(self):
        """Returns a list with descriptions of all tokens in this tokenset"""
        return [dict(cid=self.identifier)]
        
    def _addChild(self, childId):
        """Register a new child token to this one"""
        assert 0

    def _delChild(self, childId):
        """Unregister a child token"""
        
    def _save(self):
        """Saves a token back to stable storage"""
        assert 0
        
    def _revoke(self):
        """Revoke this token"""
        assert 0
        
    def _getOwner(self):
        """Is the current carrier the owner of this token?"""
        return False
        
    def _setOwner(self, newOwner):
        """Set new owner of this token"""
        assert 0

    def _getObject(self):
        """Returns the object to which this token pertains"""
        return None
        
    def _removeToken(self, tokenId):
        """Remove token tokenId from this set"""
        assert 0
        
    def addToHeadersFor(self, headers, url):
        """Add this token to the (http request) headers if it has an external representation for this destination"""
        pass

    def addToHeadersAsOTP(self, headers):
        """Add this token to the (http request) headers in one-time-password form, for internal Igor use only"""
        otp = singleton.produceOTPForToken(self)
        headers['Authorization'] = 'Basic ' + base64.b64encode(otp)
        
class IgorAccessToken(BaseAccessToken):
    """A token without an external representation that allows everything everywhere.
    To be used sparingly by Igor itself."""

    def __init__(self):
        self.identifier = '*SUPER*'
        
    def _allows(self, operation, accessChecker):
        if not operation in NORMAL_OPERATIONS:
            if DEBUG: print 'access: %s %s: not allowed by supertoken' % (operation, accessChecker.destination)
        if DEBUG: print 'access: %s %s: allowed by supertoken' % (operation, accessChecker.destination)
        return True

    def _allowsDelegation(self, path, rights, aud=None):
        """Internal method - return True, the supertoken is the root of all tokens"""
        return False

class AccessToken(BaseAccessToken):
    """An access token (or set of tokens) that can be carried by a request"""

    def __init__(self, content, defaultIdentifier=None, owner=None):
        BaseAccessToken.__init__(self)
        self.owner = owner
        self.content = dict(content)
        toDelete = []
        for k in self.content:
            if not self.content[k]:
                toDelete.append(k)
        for k in toDelete:
            del self.content[k]
        #
        # Determine identifier
        #
        if defaultIdentifier == None:
            defaultIdentifier = 'no-id-%x' % id(self)
        self.identifier = content.get('cid', defaultIdentifier)
        singleton._registerTokenWithIdentifier(self.identifier, self)
        #
        # Check whether this capability is meant for this igor (no aud or aud matches our URL)
        #
        if 'aud' in content:
            audience = content['aud']
            ourUrl = singleton.getSelfAudience()
            self.validForSelf = (audience == ourUrl)
            if DEBUG: print 'access: <aud> matches: %s' % self.validForSelf
        else:
            self.validForSelf = True
        if DEBUG:  print 'access: Created:', repr(self)
        
    def __repr__(self):
        return "%s(0x%x, %s)" % (self.__class__.__name__, id(self), repr(self.content))
        
    def _hasExternalRepresentationFor(self, url):
        if not 'iss' in self.content:
            return False
        if not 'aud' in self.content:
            return False
        if url.startswith(self.content['aud']):
            return True
        p = urlparse.urlparse(url)
        if p.netloc == self.content['aud']:
            return True
        return False

    def _getExternalContent(self):
        rv = dict(cid=self.identifier)
        for key in ['obj', 'iss', 'aud', 'sub', 'nvb', 'nva', 'get', 'put', 'post', 'delete']:
            if key in self.content:
                rv[key] = self.content[key]
        return rv
        
    def _getExternalRepresentation(self):
        tokenContent = self._getExternalContent()
        externalRepresentation = singleton._encodeOutgoingData(tokenContent)
        return externalRepresentation
        
    def _allows(self, operation, accessChecker):
        # First check this this capability is for us.
        if not self.validForSelf:
            if DEBUG: print 'access: Not for this Igor: AccessToken %s' % self
            return False
        cascadingRule = self.content.get(operation)
        if not cascadingRule:
            if DEBUG: print 'access: %s %s: no %s access allowed by AccessToken %s' % (operation, accessChecker.destination, operation, self)
            return False
        path = self.content.get('obj')
        if not path:
            if DEBUG: print 'access: %s %s: no path-based access allowed by AccessToken %s' % (operation, accessChecker.destination, self)
            return False
        dest = accessChecker.destination
        destHead = dest[:len(path)]
        destTail = dest[len(path):]
        if cascadingRule == 'self':
            if dest != path:
                if DEBUG: print 'access: %s %s: does not match cascade=%s for path=%s' % (operation, dest, cascadingRule, path)
                return False
        elif cascadingRule == 'descendant-or-self':
            if destHead != path or destTail[:1] not in ('', '/'):
                if DEBUG: print 'access: %s %s: does not match cascade=%s for path=%s' % (operation, dest, cascadingRule, path)
                return False
        elif cascadingRule == 'descendant':
            if destHead != path or destTail[:1] != '/':
                if DEBUG: print 'access: %s %s: does not match cascade=%s for path=%s' % (operation, dest, cascadingRule, path)
                return False
        elif cascadingRule == 'child':
            if destHead != path or destTail[:1] != '/' or destTail.count('/') != 1:
                if DEBUG: print 'access: %s %s: does not match cascade=%s for path=%s' % (operation, dest, cascadingRule, path)
                return False
        else:
            raise AccessControlError('Capability has unknown cascading rule %s for operation %s' % (cascadingRule, operation))
        if DEBUG: print 'access: %s %s: allowed by AccessToken %s' % (operation, accessChecker.destination, self)
        return True

    def _allowsDelegation(self, newPath, newRights, aud=None):
        """Internal method - return True if the given path/rights are a subset of this token, and if this token can be delegated"""
        # Check whether this token can be delegated
        canDelegate = self.content.get('delegate')
        if not canDelegate:
            if DEBUG_DELEGATION: print 'access: delegate %s: no delegation right on AccessToken %s' % (newPath, self)
            return False
        # Check whether this is the external-supertoken and whether we want to create a new external token
        print 'xxxjack allowsDelegation', self, 'canDelegate', canDelegate, 'aud', aud
        if canDelegate == 'external':
            if aud and aud != singleton.getSelfAudience():
                # xxxjack no further checks for external tokens. This may need refining.
                return True
            else:
                return False
        else:
            if aud and aud != singleton.getSelfAudience():
                return False
        # Check whether the path is contained in our path
        path = self.content.get('obj')
        if not path:
            if DEBUG_DELEGATION: print 'access: delegate %s: no path-based access allowed by AccessToken %s' % (newPath, self)
            return False
        subPath = newPath[len(path):]
        if not newPath.startswith(path) or not subPath[:1] in ('', '/'):
            if DEBUG_DELEGATION: print 'access: delegate %s: path not contained within path for AccessToken %s' % (newPath, self)
            return False
        newIsSelf = subPath == ''
        newIsChild = subPath.count('/') == 1
        # Check that the requested rights match
        for operation, newCascadingRule in newRights.items():
            if not newCascadingRule:
                # If we don't want this operation it is always okay.
                continue
            oldCascadingRule = self.content.get(operation)
            if not oldCascadingRule:
                # If the operation isn't allowed at all it's definitely not okay.
                if DEBUG_DELEGATION: print 'access: delegate %s: no %s access allowed by AccessToken %s' % (newPath, operation, self)
                return False
            if newIsSelf:
                if not newCascadingRule in CASCADING_RULES_IMPLIED.get(oldCascadingRule, {}):
                    if DEBUG_DELEGATION: print 'access: delegate %s: %s=%s not allowed by %s=%s for AccessToken %s' % (newPath, operation, newCascadingRule, operation, oldCascadingRule, self)
                    return False
            elif newIsChild:
                # xxxjack for now only allow if original rule includes all descendants
                if not oldCascadingRule in ('descendant', 'descendant-or-self'):
                    if DEBUG_DELEGATION: print 'access: delegate %s: %s=%s not allowd by %s=%s for AccessToken %s (xxxjack temp)' % (newPath, operation, newCascadingRule, operation, oldCascadingRule, self)
{'fraction_non_alphanumeric': 0.09320526503023835, 'fraction_numerical': 0.0016897901102810387, 'mean_word_length': 2.3307168246445498, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2727688', 'n_tokens_mistral': 6021, 'n_tokens_neox': 5869, 'n_words': 2190}