<commit_before>#
# Copyright (c) 2004-2005 rPath, Inc.
#
# This program is distributed under the terms of the Common Public License,
# version 1.0. A copy of this license should have been distributed with this
# source file in a file called LICENSE. If it is not present, the license
# is always available at http://www.opensource.org/licenses/cpl.php.
#
# This program is distributed in the hope that it will be useful, but
# without any waranty; without even the implied warranty of merchantability
# or fitness for a particular purpose. See the Common Public License for
# full details.
#
import md5
import os
import re
import sys

from conary import sqlite3
from conary.repository import errors
from conary.repository.netrepos import schema
from conary.lib.tracelog import logMe

# FIXME: remove these compatibilty error classes later
UserAlreadyExists = errors.UserAlreadyExists
GroupAlreadyExists = errors.GroupAlreadyExists

class NetworkAuthorization:

    def getAuthGroups(self, cu, authToken):
        logMe(authToken[0], authToken[2], authToken[3])
        # Find what group this user belongs to

        # anonymous users should come through as anonymous, not None
        assert(authToken[0])
        cu.execute("""SELECT salt, password, userGroupId 
                            FROM 
                                Users, UserGroupMembers
                            WHERE
                                user = ? AND
                                Users.userId = UserGroupMembers.userId""",
                   authToken[0])

        groupsFromUser = [ x for x in cu ]
        
        if groupsFromUser:
            # each user can only appear once (by constraint), so we only
            # need to validate the password once
            if not self.checkPassword(groupsFromUser[0][0], 
                                      groupsFromUser[0][1],
                                      authToken[1]):
                raise errors.InsufficientPermission

            groupsFromUser = [ x[2] for x in groupsFromUser ]

        if authToken[2] is not None:
            # look up entitlements
            cu.execute("""
                  SELECT userGroupId 
                        FROM EntitlementGroups, Entitlements 
                  WHERE 
                        entGroup=? AND 
                        entitlement=? AND
                        EntitlementGroups.entGroupId == Entitlements.entGroupId
                """, authToken[2], authToken[3])
            groupsFromUser += [ x[0] for x in cu ]

        return groupsFromUser

    def check(self, authToken, write = False, admin = False, label = None, 
              trove = None):
        logMe(2, authToken[0], authToken[1], authToken[2], write, admin, 
              label, trove)

        if label and label.getHost() != self.name:
            raise errors.RepositoryMismatch

        if not authToken[0]:
            return False

        cu = self.db.cursor()

        groupIds = self.getAuthGroups(cu, authToken)

        stmt = """
        select Items.item
        from Permissions join items using (itemId)
        """

        where = ["Permissions.userGroupId IN (%s)" % 
                    ",".join("%d" % x for x in groupIds) ]
        params = []

        if label:
            where.append("""
            (
            Permissions.labelId = 0 OR
            Permissions.labelId in
                ( select labelId from Labels where Labels.label = ? )
            )
            """)
            params.append(label.asString())

        if write:
            where.append("Permissions.write=1")<commit_msg>check() should return true/false, not raise an error<commit_after>#
# Copyright (c) 2004-2005 rPath, Inc.
#
# This program is distributed under the terms of the Common Public License,
# version 1.0. A copy of this license should have been distributed with this
# source file in a file called LICENSE. If it is not present, the license
# is always available at http://www.opensource.org/licenses/cpl.php.
#
# This program is distributed in the hope that it will be useful, but
# without any waranty; without even the implied warranty of merchantability
# or fitness for a particular purpose. See the Common Public License for
# full details.
#
import md5
import os
import re
import sys

from conary import sqlite3
from conary.repository import errors
from conary.repository.netrepos import schema
from conary.lib.tracelog import logMe

# FIXME: remove these compatibilty error classes later
UserAlreadyExists = errors.UserAlreadyExists
GroupAlreadyExists = errors.GroupAlreadyExists

class NetworkAuthorization:

    def getAuthGroups(self, cu, authToken):
        logMe(authToken[0], authToken[2], authToken[3])
        # Find what group this user belongs to

        # anonymous users should come through as anonymous, not None
        assert(authToken[0])
        cu.execute("""SELECT salt, password, userGroupId 
                            FROM 
                                Users, UserGroupMembers
                            WHERE
                                user = ? AND
                                Users.userId = UserGroupMembers.userId""",
                   authToken[0])

        groupsFromUser = [ x for x in cu ]
        
        if groupsFromUser:
            # each user can only appear once (by constraint), so we only
            # need to validate the password once
            if not self.checkPassword(groupsFromUser[0][0], 
                                      groupsFromUser[0][1],
                                      authToken[1]):
                raise errors.InsufficientPermission

            groupsFromUser = [ x[2] for x in groupsFromUser ]

        if authToken[2] is not None:
            # look up entitlements
            cu.execute("""
                  SELECT userGroupId 
                        FROM EntitlementGroups, Entitlements 
                  WHERE 
                        entGroup=? AND 
                        entitlement=? AND
                        EntitlementGroups.entGroupId == Entitlements.entGroupId
                """, authToken[2], authToken[3])
            groupsFromUser += [ x[0] for x in cu ]

        return groupsFromUser

    def check(self, authToken, write = False, admin = False, label = None, 
              trove = None):
        logMe(2, authToken[0], authToken[1], authToken[2], write, admin, 
              label, trove)

        if label and label.getHost() != self.name:
            raise errors.RepositoryMismatch

        if not authToken[0]:
            return False

        cu = self.db.cursor()

        try:
            groupIds = self.getAuthGroups(cu, authToken)
        except errors.InsufficientPermission:
            return False

        stmt = """
        select Items.item
        from Permissions join items using (itemId)
        """

        where = ["Permissions.userGroupId IN (%s)" % 
                    ",".join("%d" % x for x in groupIds) ]
        params = []

        if label:
            where.append("""
            (
            Permissions.labelId = 0 OR
            Permissions.labelId in
                ( select labelId from Labels where Labels.label = ? )
            )
            """)
            params.append(label.asString())

        if write:
            where.append("Permissions.write=1")
{'fraction_non_alphanumeric': 0.07553612894413332, 'fraction_numerical': 0.009288348586258708, 'mean_word_length': 1.7495306045812993, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '113562', 'n_tokens_mistral': 1895, 'n_tokens_neox': 1780, 'n_words': 619}