<s> import os <EOL> from grr . lib import access_control <EOL> from grr . lib import aff4 <EOL> from grr . lib import config_lib <EOL> from grr . lib import data_store <EOL> from grr . lib import email_alerts <EOL> from grr . lib import flags <EOL> from grr . lib import flow <EOL> from grr . lib import hunts <EOL> from grr . lib import rdfvalue <EOL> from grr . lib import test_lib <EOL> from grr . lib import utils <EOL> from grr . lib . aff4_objects import security <EOL> from grr . lib . aff4_objects import user_managers <EOL> from grr . lib . aff4_objects import users <EOL> from grr . lib . authorization import client_approval_auth <EOL> from grr . lib . rdfvalues import aff4_rdfvalues <EOL> from grr . lib . rdfvalues import client as rdf_client <EOL> class GRRUserTest ( test_lib . AFF4ObjectTest ) : <EOL> def testUserPasswords ( self ) : <EOL> with aff4 . FACTORY . Create ( "<STR_LIT>" , "<STR_LIT>" , <EOL> token = self . token ) as user : <EOL> user . SetPassword ( "<STR_LIT:hello>" ) <EOL> user = aff4 . FACTORY . Open ( user . urn , token = self . token ) <EOL> self . assertFalse ( user . CheckPassword ( "<STR_LIT>" ) ) <EOL> self . assertTrue ( user . CheckPassword ( "<STR_LIT:hello>" ) ) <EOL> def testLabels ( self ) : <EOL> with aff4 . FACTORY . Create ( "<STR_LIT>" , "<STR_LIT>" , <EOL> token = self . token ) as user : <EOL> user . SetLabels ( "<STR_LIT:hello>" , "<STR_LIT>" , owner = "<STR_LIT>" ) <EOL> user = aff4 . FACTORY . Open ( user . urn , token = self . token ) <EOL> self . assertListEqual ( [ "<STR_LIT:hello>" , "<STR_LIT>" ] , user . GetLabelsNames ( ) ) <EOL> def testBackwardsCompatibility ( self ) : <EOL> """<STR_LIT>""" <EOL> password = users . CryptedPassword ( ) <EOL> password . _value = "<STR_LIT>" <EOL> self . assertFalse ( password . CheckPassword ( "<STR_LIT>" ) ) <EOL> self . assertTrue ( password . CheckPassword ( "<STR_LIT:hello>" ) ) <EOL> class CheckAccessHelperTest ( test_lib . GRRBaseTest ) : <EOL> def setUp ( self ) : <EOL> super ( CheckAccessHelperTest , self ) . setUp ( ) <EOL> self . helper = user_managers . CheckAccessHelper ( "<STR_LIT:test>" ) <EOL> self . subject = rdfvalue . RDFURN ( "<STR_LIT>" ) <EOL> def testReturnsFalseByDefault ( self ) : <EOL> self . assertRaises ( access_control . UnauthorizedAccess , <EOL> self . helper . CheckAccess , self . subject , self . token ) <EOL> def testReturnsFalseOnFailedMatch ( self ) : <EOL> self . helper . Allow ( "<STR_LIT>" ) <EOL> self . assertRaises ( access_control . UnauthorizedAccess , <EOL> self . helper . CheckAccess , self . subject , self . token ) <EOL> def testReturnsTrueOnMatch ( self ) : <EOL> self . helper . Allow ( "<STR_LIT>" ) <EOL> self . assertTrue ( self . helper . CheckAccess ( self . subject , self . token ) ) <EOL> def testReturnsTrueIfOneMatchFails1 ( self ) : <EOL> self . helper . Allow ( "<STR_LIT>" ) <EOL> self . helper . Allow ( "<STR_LIT>" ) <EOL> self . assertTrue ( self . helper . CheckAccess ( self . subject , self . token ) ) <EOL> def testReturnsTrueIfOneMatchFails2 ( self ) : <EOL> self . helper . Allow ( "<STR_LIT>" ) <EOL> self . helper . Allow ( "<STR_LIT>" ) <EOL> self . assertTrue ( self . helper . CheckAccess ( self . subject , self . token ) ) <EOL> def testFnmatchFormatIsUsedByDefault1 ( self ) : <EOL> self . helper . Allow ( "<STR_LIT>" ) <EOL> self . assertTrue ( self . helper . CheckAccess ( self . subject , self . token ) ) <EOL> def testFnmatchFormatIsUsedByDefault2 ( self ) : <EOL> self . helper . Allow ( "<STR_LIT>" ) <EOL> self . assertTrue ( self . helper . CheckAccess ( self . subject , self . token ) ) <EOL> def testFnmatchPatternCorrectlyMatchesFilesBelowDirectory ( self ) : <EOL> self . helper . Allow ( "<STR_LIT>" ) <EOL> self . assertTrue ( self . helper . CheckAccess ( self . subject , self . token ) ) <EOL> self . assertRaises ( access_control . UnauthorizedAccess , <EOL> self . helper . CheckAccess , <EOL> rdfvalue . RDFURN ( "<STR_LIT>" ) , self . token ) <EOL> def testCustomCheckWorksCorrectly ( self ) : <EOL> def CustomCheck ( unused_subject , unused_token ) : <EOL> return True <EOL> self . helper . Allow ( "<STR_LIT>" , CustomCheck ) <EOL> self . assertTrue ( self . helper . CheckAccess ( self . subject , self . token ) ) <EOL> def testCustomCheckFailsCorrectly ( self ) : <EOL> def CustomCheck ( unused_subject , unused_token ) : <EOL> raise access_control . UnauthorizedAccess ( "<STR_LIT>" ) <EOL> self . helper . Allow ( "<STR_LIT>" , CustomCheck ) <EOL> self . assertRaises ( access_control . UnauthorizedAccess , <EOL> self . helper . CheckAccess , self . subject , self . token ) <EOL> def testCustomCheckAcceptsAdditionalArguments ( self ) : <EOL> def CustomCheck ( subject , unused_token , another_subject ) : <EOL> if subject == another_subject : <EOL> return True <EOL> else : <EOL> raise access_control . UnauthorizedAccess ( "<STR_LIT>" ) <EOL> self . helper . Allow ( "<STR_LIT>" , CustomCheck , self . subject ) <EOL> self . assertRaises ( access_control . UnauthorizedAccess , <EOL> self . helper . CheckAccess , <EOL> rdfvalue . RDFURN ( "<STR_LIT>" ) , <EOL> self . token ) <EOL> self . assertTrue ( self . helper . CheckAccess ( self . subject , self . token ) ) <EOL> class AdminOnlyFlow ( flow . GRRFlow ) : <EOL> AUTHORIZED_LABELS = [ "<STR_LIT>" ] <EOL> category = "<STR_LIT>" <EOL> class BasicAccessControlManagerTest ( test_lib . GRRBaseTest ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> super ( BasicAccessControlManagerTest , self ) . setUp ( ) <EOL> self . access_manager = user_managers . BasicAccessControlManager ( ) <EOL> def testUserWithoutAuthorizedLabelsCanNotStartFlow ( self ) : <EOL> self . CreateUser ( "<STR_LIT>" ) <EOL> nonadmin_token = access_control . ACLToken ( username = "<STR_LIT>" , <EOL> reason = "<STR_LIT>" ) <EOL> with self . assertRaises ( access_control . UnauthorizedAccess ) : <EOL> self . access_manager . CheckIfCanStartFlow ( nonadmin_token , <EOL> AdminOnlyFlow . __name__ , <EOL> with_client_id = False ) <EOL> with self . assertRaises ( access_control . UnauthorizedAccess ) : <EOL> self . access_manager . CheckIfCanStartFlow ( nonadmin_token , <EOL> AdminOnlyFlow . __name__ , <EOL> with_client_id = True ) <EOL> def testUserWithAuthorizedLabelsCanStartFlow ( self ) : <EOL> self . CreateAdminUser ( "<STR_LIT>" ) <EOL> admin_token = access_control . ACLToken ( username = "<STR_LIT>" , <EOL> reason = "<STR_LIT>" ) <EOL> self . access_manager . CheckIfCanStartFlow ( admin_token , <EOL> AdminOnlyFlow . __name__ , <EOL> with_client_id = False ) <EOL> self . access_manager . CheckIfCanStartFlow ( admin_token , <EOL> AdminOnlyFlow . __name__ , <EOL> with_client_id = True ) <EOL> class FullAccessControlManagerTest ( test_lib . GRRBaseTest ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> super ( FullAccessControlManagerTest , self ) . setUp ( ) <EOL> self . access_manager = user_managers . FullAccessControlManager ( ) <EOL> def Ok ( self , subject , access = "<STR_LIT:r>" ) : <EOL> self . assertTrue ( <EOL> self . access_manager . CheckDataStoreAccess ( self . token , [ subject ] , access ) ) <EOL> def NotOk ( self , subject , access = "<STR_LIT:r>" ) : <EOL> self . assertRaises ( <EOL> access_control . UnauthorizedAccess , <EOL> self . access_manager . CheckDataStoreAccess , <EOL> self . token , [ subject ] , access ) <EOL> def testReadSomePaths ( self ) : <EOL> """<STR_LIT>""" <EOL> access = "<STR_LIT:r>" <EOL> self . Ok ( "<STR_LIT>" , access ) <EOL> self . Ok ( "<STR_LIT>" , access ) <EOL> self . NotOk ( "<STR_LIT>" , access ) <EOL> self . Ok ( "<STR_LIT>" , access ) <EOL> self . Ok ( "<STR_LIT>" , access ) <EOL> self . Ok ( "<STR_LIT>" , access ) <EOL> self . Ok ( "<STR_LIT>" , access ) <EOL> self . Ok ( "<STR_LIT>" , access ) <EOL> self . Ok ( "<STR_LIT>" , access ) <EOL> self . Ok ( "<STR_LIT>" , access ) <EOL> self . Ok ( "<STR_LIT>" , access ) <EOL> self . Ok ( "<STR_LIT>" , access ) <EOL> self . Ok ( "<STR_LIT>" , access ) <EOL> self . Ok ( "<STR_LIT>" , access ) <EOL> self . Ok ( "<STR_LIT>" , access ) <EOL> self . Ok ( "<STR_LIT>" , access ) <EOL> self . Ok ( "<STR_LIT>" , access ) <EOL> self . Ok ( "<STR_LIT>" , access ) <EOL> self . Ok ( "<STR_LIT>" , access ) <EOL> self . Ok ( "<STR_LIT>" , access ) <EOL> self . Ok ( "<STR_LIT>" , access ) <EOL> self . Ok ( "<STR_LIT>" , access ) <EOL> self . Ok ( "<STR_LIT>" , access ) <EOL> self . Ok ( "<STR_LIT>" , access ) <EOL> self . Ok ( "<STR_LIT>" , access ) <EOL> self . Ok ( "<STR_LIT>" , access ) <EOL> self . Ok ( "<STR_LIT>" , access ) <EOL> self . Ok ( "<STR_LIT>" , access ) <EOL> self . Ok ( "<STR_LIT>" , access ) <EOL> self . NotOk ( "<STR_LIT>" , access ) <EOL> self . NotOk ( "<STR_LIT>" , access ) <EOL> self . Ok ( "<STR_LIT>" , access ) <EOL> self . Ok ( "<STR_LIT>" , access ) <EOL> def testQuerySomePaths ( self ) : <EOL> """<STR_LIT>""" <EOL> access = "<STR_LIT>" <EOL> self . NotOk ( "<STR_LIT>" , access ) <EOL> self . NotOk ( "<STR_LIT>" , access ) <EOL> self . NotOk ( "<STR_LIT>" , access ) <EOL> self . NotOk ( "<STR_LIT>" , access ) <EOL> self . NotOk ( "<STR_LIT>" , access ) <EOL> self . NotOk ( "<STR_LIT>" , access ) <EOL> self . Ok ( "<STR_LIT>" + "<STR_LIT:a>" * <NUM_LIT:64> , access ) <EOL> self . Ok ( "<STR_LIT>" , access ) <EOL> self . Ok ( "<STR_LIT>" , access ) <EOL> self . NotOk ( "<STR_LIT>" , access ) <EOL> self . Ok ( "<STR_LIT>" , access ) <EOL> self . Ok ( "<STR_LIT>" , access ) <EOL> self . Ok ( "<STR_LIT>" , access ) <EOL> self . NotOk ( "<STR_LIT>" , access ) <EOL> self . Ok ( "<STR_LIT>" , access ) <EOL> self . Ok ( "<STR_LIT>" , access ) <EOL> self . Ok ( "<STR_LIT>" , access ) <EOL> self . Ok ( "<STR_LIT>" , access ) <EOL> self . Ok ( "<STR_LIT>" , access ) <EOL> self . Ok ( "<STR_LIT>" , access ) <EOL> self . NotOk ( "<STR_LIT>" , access ) <EOL> self . NotOk ( "<STR_LIT>" , access ) <EOL> self . Ok ( "<STR_LIT>" , access ) <EOL> self . Ok ( "<STR_LIT>" , access ) <EOL> self . NotOk ( "<STR_LIT>" , access ) <EOL> self . NotOk ( "<STR_LIT>" , access ) <EOL> self . NotOk ( "<STR_LIT>" , access ) <EOL> def testSupervisorCanDoAnything ( self ) : <EOL> token = access_control . ACLToken ( username = "<STR_LIT>" , supervisor = True ) <EOL> self . assertTrue ( self . access_manager . CheckClientAccess ( <EOL> token , "<STR_LIT>" ) ) <EOL> self . assertTrue ( self . access_manager . CheckHuntAccess ( <EOL> token , "<STR_LIT>" ) ) <EOL> self . assertTrue ( self . access_manager . CheckCronJobAccess ( <EOL> token , "<STR_LIT>" ) ) <EOL> self . assertTrue ( self . access_manager . CheckIfCanStartFlow ( <EOL> token , "<STR_LIT>" , with_client_id = True ) ) <EOL> self . assertTrue ( self . access_manager . CheckIfCanStartFlow ( <EOL> token , "<STR_LIT>" , with_client_id = False ) ) <EOL> self . assertTrue ( self . access_manager . CheckDataStoreAccess ( <EOL> token , [ "<STR_LIT>" ] , requested_access = "<STR_LIT:w>" ) ) <EOL> def testEmptySubjectShouldRaise ( self ) : <EOL> token = access_control . ACLToken ( username = "<STR_LIT>" ) <EOL> with self . assertRaises ( ValueError ) : <EOL> self . access_manager . CheckClientAccess ( token , "<STR_LIT>" ) <EOL> with self . assertRaises ( ValueError ) : <EOL> self . access_manager . CheckHuntAccess ( token , "<STR_LIT>" ) <EOL> with self . assertRaises ( ValueError ) : <EOL> self . access_manager . CheckCronJobAccess ( token , "<STR_LIT>" ) <EOL> with self . assertRaises ( ValueError ) : <EOL> self . access_manager . CheckDataStoreAccess ( token , [ "<STR_LIT>" ] , <EOL> requested_access = "<STR_LIT:r>" ) <EOL> def testCheckIfCanStartFlowReturnsTrueForClientFlowOnClient ( self ) : <EOL> self . assertTrue ( self . access_manager . CheckIfCanStartFlow ( <EOL> self . token , ClientFlowWithCategory . __name__ , with_client_id = True ) ) <EOL> def testCheckIfCanStartFlowRaisesForClientFlowWithoutCategoryOnClient ( self ) : <EOL> with self . assertRaises ( access_control . UnauthorizedAccess ) : <EOL> self . access_manager . CheckIfCanStartFlow ( <EOL> self . token , ClientFlowWithoutCategory . __name__ , with_client_id = True ) <EOL> def testCheckIfCanStartFlowReturnsTrueForNotEnforcedFlowOnClient ( self ) : <EOL> self . assertTrue ( self . access_manager . CheckIfCanStartFlow ( <EOL> self . token , NotEnforcedFlow . __name__ , with_client_id = True ) ) <EOL> def testCheckIfCanStartFlowRaisesForClientFlowAsGlobal ( self ) : <EOL> with self . assertRaises ( access_control . UnauthorizedAccess ) : <EOL> self . access_manager . CheckIfCanStartFlow ( <EOL> self . token , ClientFlowWithCategory . __name__ , with_client_id = False ) <EOL> def testCheckIfCanStartFlowRaisesForGlobalFlowWithoutCategoryAsGlobal ( self ) : <EOL> with self . assertRaises ( access_control . UnauthorizedAccess ) : <EOL> self . access_manager . CheckIfCanStartFlow ( <EOL> self . token , GlobalFlowWithoutCategory . __name__ , with_client_id = False ) <EOL> def testCheckIfCanStartFlowReturnsTrueForGlobalFlowWithCategoryAsGlobal ( self ) : <EOL> self . assertTrue ( self . access_manager . CheckIfCanStartFlow ( <EOL> self . token , GlobalFlowWithCategory . __name__ , with_client_id = False ) ) <EOL> def testNoReasonShouldSearchForApprovals ( self ) : <EOL> token_without_reason = access_control . ACLToken ( username = "<STR_LIT>" ) <EOL> token_with_reason = access_control . ACLToken ( username = "<STR_LIT>" , <EOL> reason = "<STR_LIT>" ) <EOL> client_id = "<STR_LIT>" <EOL> self . RequestAndGrantClientApproval ( client_id , token = token_with_reason ) <EOL> self . access_manager . CheckClientAccess ( token_without_reason , client_id ) <EOL> self . assertEqual ( token_without_reason . reason , "<STR_LIT>" ) <EOL> class ValidateTokenTest ( test_lib . GRRBaseTest ) : <EOL> """<STR_LIT>""" <EOL> def testTokenWithUsernameAndReasonIsValid ( self ) : <EOL> token = access_control . ACLToken ( username = "<STR_LIT:test>" , reason = "<STR_LIT>" ) <EOL> user_managers . ValidateToken ( token , "<STR_LIT>" ) <EOL> def testNoneTokenIsNotValid ( self ) : <EOL> with self . assertRaises ( access_control . UnauthorizedAccess ) : <EOL> user_managers . ValidateToken ( None , "<STR_LIT>" ) <EOL> def testTokenWithoutUsernameIsNotValid ( self ) : <EOL> token = access_control . ACLToken ( reason = "<STR_LIT>" ) <EOL> with self . assertRaises ( access_control . UnauthorizedAccess ) : <EOL> user_managers . ValidateToken ( token , "<STR_LIT>" ) <EOL> class ClientFlowWithoutCategory ( flow . GRRFlow ) : <EOL> pass <EOL> class ClientFlowWithCategory ( flow . GRRFlow ) : <EOL> category = "<STR_LIT>" <EOL> class NotEnforcedFlow ( flow . GRRFlow ) : <EOL> ACL_ENFORCED = False <EOL> class GlobalFlowWithoutCategory ( flow . GRRGlobalFlow ) : <EOL> pass <EOL> class GlobalFlowWithCategory ( flow . GRRGlobalFlow ) : <EOL> category = "<STR_LIT>" <EOL> class FullAccessControlManagerIntegrationTest ( test_lib . GRRBaseTest ) : <EOL> """<STR_LIT>""" <EOL> install_mock_acl = False <EOL> def setUp ( self ) : <EOL> super ( FullAccessControlManagerIntegrationTest , self ) . setUp ( ) <EOL> data_store . DB . security_manager = access_control . FullAccessControlManager ( ) <EOL> def ACLChecksDisabled ( self ) : <EOL> return test_lib . ACLChecksDisabledContextManager ( ) <EOL> def RevokeClientApproval ( self , client_id , token , remove_from_cache = True ) : <EOL> approval_urn = aff4 . ROOT_URN . Add ( "<STR_LIT>" ) . Add ( client_id ) . Add ( <EOL> token . username ) . Add ( utils . EncodeReasonString ( token . reason ) ) <EOL> with aff4 . FACTORY . Open ( approval_urn , mode = "<STR_LIT>" , <EOL> token = self . token . SetUID ( ) ) as approval_request : <EOL> approval_request . DeleteAttribute ( approval_request . Schema . APPROVER ) <EOL> if remove_from_cache : <EOL> data_store . DB . security_manager . acl_cache . ExpireObject ( approval_urn ) <EOL> def CreateHuntApproval ( self , hunt_urn , token , admin = False ) : <EOL> approval_urn = aff4 . ROOT_URN . Add ( "<STR_LIT>" ) . Add ( hunt_urn . Path ( ) ) . Add ( <EOL> token . username ) . Add ( utils . EncodeReasonString ( token . reason ) ) <EOL> with aff4 . FACTORY . Create ( <EOL> approval_urn , security . HuntApproval . __name__ , mode = "<STR_LIT>" , <EOL> token = self . token . SetUID ( ) ) as approval_request : <EOL> approval_request . AddAttribute ( <EOL> approval_request . Schema . APPROVER ( "<STR_LIT>" ) ) <EOL> approval_request . AddAttribute ( <EOL> approval_request . Schema . APPROVER ( "<STR_LIT>" ) ) <EOL> if admin : <EOL> self . CreateAdminUser ( "<STR_LIT>" ) <EOL> def CreateSampleHunt ( self ) : <EOL> """<STR_LIT>""" <EOL> with hunts . GRRHunt . StartHunt ( <EOL> hunt_name = "<STR_LIT>" , token = self . token . SetUID ( ) ) as hunt : <EOL> return hunt . session_id <EOL> def testSimpleAccess ( self ) : <EOL> """<STR_LIT>""" <EOL> client_urn = rdf_client . ClientURN ( "<STR_LIT>" % <NUM_LIT:0> ) <EOL> for urn , mode in [ ( "<STR_LIT>" , "<STR_LIT:r>" ) , <EOL> ( "<STR_LIT>" , "<STR_LIT:r>" ) , <EOL> ( "<STR_LIT>" , "<STR_LIT>" ) , <EOL> ( client_urn , "<STR_LIT:r>" ) ] : <EOL> self . assertRaises ( access_control . UnauthorizedAccess , aff4 . FACTORY . Open , <EOL> urn , mode = mode ) <EOL> for urn , mode in [ ( "<STR_LIT>" , "<STR_LIT>" ) , <EOL> ( client_urn , "<STR_LIT>" ) ] : <EOL> fd = aff4 . FACTORY . Open ( urn , mode = mode , token = self . token ) <EOL> fd . _dirty = True <EOL> self . assertRaises ( access_control . UnauthorizedAccess , fd . Close ) <EOL> for urn , mode in [ ( client_urn . Add ( "<STR_LIT>" ) . Add ( "<STR_LIT>" ) , "<STR_LIT:r>" ) , <EOL> ( client_urn . Add ( "<STR_LIT>" ) , "<STR_LIT:r>" ) ] : <EOL> self . assertRaises ( access_control . UnauthorizedAccess , aff4 . FACTORY . Open , <EOL> urn , mode = mode ) <EOL> self . assertRaises ( access_control . UnauthorizedAccess , aff4 . FACTORY . Open , <EOL> urn , mode = mode , token = self . token ) <EOL> def testSupervisorToken ( self ) : <EOL> """<STR_LIT>""" <EOL> urn = rdf_client . ClientURN ( "<STR_LIT>" % <NUM_LIT:0> ) . Add ( "<STR_LIT>" ) <EOL> self . assertRaises ( access_control . UnauthorizedAccess , aff4 . FACTORY . Open , urn ) <EOL> super_token = access_control . ACLToken ( username = "<STR_LIT:test>" ) <EOL> super_token . supervisor = True <EOL> aff4 . FACTORY . Open ( urn , mode = "<STR_LIT>" , token = super_token ) <EOL> def testExpiredTokens ( self ) : <EOL> """<STR_LIT>""" <EOL> urn = rdf_client . ClientURN ( "<STR_LIT>" % <NUM_LIT:0> ) . Add ( "<STR_LIT>" ) <EOL> self . assertRaises ( access_control . UnauthorizedAccess , aff4 . FACTORY . Open , urn ) <EOL> with test_lib . FakeTime ( <NUM_LIT:100> ) : <EOL> super_token = access_control . ACLToken ( username = "<STR_LIT:test>" , expiry = <NUM_LIT> ) <EOL> super_token . supervisor = True <EOL> aff4 . FACTORY . Open ( urn , mode = "<STR_LIT>" , token = super_token ) <EOL> with test_lib . FakeTime ( <NUM_LIT:200> ) : <EOL> self . assertRaises ( access_control . ExpiryError , aff4 . FACTORY . Open , urn , <EOL> token = super_token , mode = "<STR_LIT>" ) <EOL> def testApprovalExpiry ( self ) : <EOL> """<STR_LIT>""" <EOL> client_id = "<STR_LIT>" % <NUM_LIT:0> <EOL> urn = rdf_client . ClientURN ( client_id ) . Add ( "<STR_LIT>" ) <EOL> token = access_control . ACLToken ( username = "<STR_LIT:test>" , reason = "<STR_LIT>" ) <EOL> self . assertRaises ( access_control . UnauthorizedAccess , aff4 . FACTORY . Open , urn , <EOL> None , "<STR_LIT>" , token ) <EOL> with test_lib . FakeTime ( <NUM_LIT> , increment = <NUM_LIT> ) : <EOL> self . RequestAndGrantClientApproval ( client_id , token ) <EOL> aff4 . FACTORY . Open ( urn , mode = "<STR_LIT>" , token = token ) <EOL> token_expiry = config_lib . CONFIG [ "<STR_LIT>" ] <EOL> with test_lib . FakeTime ( <NUM_LIT> + token_expiry - <NUM_LIT> ) : <EOL> aff4 . FACTORY . Open ( urn , mode = "<STR_LIT>" , token = token ) <EOL> with test_lib . FakeTime ( <NUM_LIT> + token_expiry + <NUM_LIT> ) : <EOL> self . assertRaises ( access_control . UnauthorizedAccess , <EOL> aff4 . FACTORY . Open , urn , None , "<STR_LIT>" , token ) <EOL> def testClientApproval ( self ) : <EOL> """<STR_LIT>""" <EOL> client_id = "<STR_LIT>" % <NUM_LIT:0> <EOL> urn = rdf_client . ClientURN ( client_id ) . Add ( "<STR_LIT>" ) <EOL> token = access_control . ACLToken ( username = "<STR_LIT:test>" , reason = "<STR_LIT>" ) <EOL> self . assertRaises ( access_control . UnauthorizedAccess , aff4 . FACTORY . Open , urn , <EOL> None , "<STR_LIT>" , token = token ) <EOL> self . RequestAndGrantClientApproval ( client_id , token ) <EOL> fd = aff4 . FACTORY . Open ( urn , None , "<STR_LIT>" , token = token ) <EOL> fd . Close ( ) <EOL> self . RevokeClientApproval ( client_id , token ) <EOL> self . assertRaises ( access_control . UnauthorizedAccess , <EOL> aff4 . FACTORY . Open , urn , None , "<STR_LIT>" , token = token ) <EOL> def testHuntApproval ( self ) : <EOL> """<STR_LIT>""" <EOL> token = access_control . ACLToken ( username = "<STR_LIT:test>" , reason = "<STR_LIT>" ) <EOL> hunt_urn = self . CreateSampleHunt ( ) <EOL> self . assertRaisesRegexp ( <EOL> access_control . UnauthorizedAccess , <EOL> "<STR_LIT>" , <EOL> flow . GRRFlow . StartFlow , <EOL> flow_name = "<STR_LIT>" , token = token , hunt_urn = hunt_urn ) <EOL> self . CreateHuntApproval ( hunt_urn , token , admin = False ) <EOL> self . assertRaisesRegexp ( <EOL> access_control . UnauthorizedAccess , <EOL> r"<STR_LIT>" , <EOL> flow . GRRFlow . StartFlow , <EOL> flow_name = "<STR_LIT>" , token = token , hunt_urn = hunt_urn ) <EOL> self . CreateHuntApproval ( hunt_urn , token , admin = True ) <EOL> flow . GRRFlow . StartFlow ( flow_name = "<STR_LIT>" , token = token , <EOL> hunt_urn = hunt_urn ) <EOL> def testUserAccess ( self ) : <EOL> """<STR_LIT>""" <EOL> token = access_control . ACLToken ( username = "<STR_LIT:test>" , reason = "<STR_LIT>" ) <EOL> urn = aff4 . ROOT_URN . Add ( "<STR_LIT>" ) <EOL> self . assertRaises ( access_control . UnauthorizedAccess , <EOL> aff4 . FACTORY . Open , urn . Add ( "<STR_LIT>" ) , None , "<STR_LIT>" , <EOL> False , token ) <EOL> aff4 . FACTORY . Open ( urn . Add ( "<STR_LIT:test>" ) , mode = "<STR_LIT>" , token = token ) <EOL> label_urn = urn . Add ( "<STR_LIT:test>" ) . Add ( "<STR_LIT>" ) <EOL> labels = aff4 . FACTORY . Open ( label_urn , mode = "<STR_LIT>" , token = token ) <EOL> l = labels . Schema . LABELS ( ) <EOL> l . AddLabel ( aff4_rdfvalues . AFF4ObjectLabel ( name = "<STR_LIT>" , owner = "<STR_LIT>" ) ) <EOL> labels . Set ( labels . Schema . LABELS , l ) <EOL> self . assertRaises ( access_control . UnauthorizedAccess , labels . Close ) <EOL> def testForemanAccess ( self ) : <EOL> """<STR_LIT>""" <EOL> token = access_control . ACLToken ( username = "<STR_LIT:test>" , reason = "<STR_LIT>" ) <EOL> self . assertRaises ( access_control . UnauthorizedAccess , <EOL> aff4 . FACTORY . Open , "<STR_LIT>" , token = token ) <EOL> super_token = access_control . ACLToken ( username = "<STR_LIT:test>" ) <EOL> super_token . supervisor = True <EOL> with aff4 . FACTORY . Create ( "<STR_LIT>" , "<STR_LIT>" , <EOL> token = super_token ) as fd : <EOL> fd . SetLabels ( "<STR_LIT>" , owner = "<STR_LIT>" ) <EOL> aff4 . FACTORY . Open ( "<STR_LIT>" , token = token ) <EOL> def testCrashesAccess ( self ) : <EOL> super_token = access_control . ACLToken ( username = "<STR_LIT:test>" ) <EOL> super_token . supervisor = True <EOL> path = rdfvalue . RDFURN ( "<STR_LIT>" ) <EOL> crashes = aff4 . FACTORY . Create ( path , "<STR_LIT>" , token = self . token ) <EOL> self . assertRaises ( access_control . UnauthorizedAccess , crashes . Close ) <EOL> crashes = aff4 . FACTORY . Create ( path , "<STR_LIT>" , <EOL> token = super_token ) <EOL> crashes . Close ( ) <EOL> crashes = aff4 . FACTORY . Open ( path , aff4_type = "<STR_LIT>" , <EOL> mode = "<STR_LIT>" , token = self . token ) <EOL> crashes . Set ( crashes . Schema . DESCRIPTION ( "<STR_LIT>" ) ) <EOL> self . assertRaises ( access_control . UnauthorizedAccess , crashes . Close ) <EOL> crashes = aff4 . FACTORY . Open ( path , aff4_type = "<STR_LIT>" , <EOL> mode = "<STR_LIT:r>" , token = self . token ) <EOL> crashes . Close ( ) <EOL> def testFlowAccess ( self ) : <EOL> """<STR_LIT>""" <EOL> token = access_control . ACLToken ( username = "<STR_LIT:test>" , reason = "<STR_LIT>" ) <EOL> client_id = "<STR_LIT>" + "<STR_LIT:a>" * <NUM_LIT:16> <EOL> self . assertRaises ( access_control . UnauthorizedAccess , flow . GRRFlow . StartFlow , <EOL> client_id = client_id , <EOL> flow_name = test_lib . SendingFlow . __name__ , <EOL> message_count = <NUM_LIT:1> , token = token ) <EOL> self . RequestAndGrantClientApproval ( client_id , token ) <EOL> sid = flow . GRRFlow . StartFlow ( <EOL> client_id = client_id , flow_name = test_lib . SendingFlow . __name__ , <EOL> message_count = <NUM_LIT:1> , token = token ) <EOL> flow_obj = aff4 . FACTORY . Open ( sid , mode = "<STR_LIT:r>" , token = token ) <EOL> flow_obj . mode = "<STR_LIT>" <EOL> state = flow_obj . Get ( flow_obj . Schema . FLOW_STATE ) <EOL> flow_obj . Set ( state ) <EOL> self . assertRaises ( access_control . UnauthorizedAccess , <EOL> flow_obj . Close ) <EOL> self . RevokeClientApproval ( client_id , token ) <EOL> self . assertRaises ( access_control . UnauthorizedAccess , <EOL> aff4 . FACTORY . Open , sid , mode = "<STR_LIT:r>" , token = token ) <EOL> self . RequestAndGrantClientApproval ( client_id , token ) <EOL> aff4 . FACTORY . Open ( sid , mode = "<STR_LIT:r>" , token = token ) <EOL> def testCaches ( self ) : <EOL> """<STR_LIT>""" <EOL> token = access_control . ACLToken ( username = "<STR_LIT:test>" , reason = "<STR_LIT>" ) <EOL> client_id = "<STR_LIT>" + "<STR_LIT:b>" * <NUM_LIT:16> <EOL> self . RequestAndGrantClientApproval ( client_id , token ) <EOL> sid = flow . GRRFlow . StartFlow ( <EOL> client_id = client_id , flow_name = test_lib . SendingFlow . __name__ , <EOL> message_count = <NUM_LIT:1> , token = token ) <EOL> aff4 . FACTORY . Open ( sid , mode = "<STR_LIT:r>" , token = token ) <EOL> aff4 . FACTORY . Flush ( ) <EOL> self . RevokeClientApproval ( client_id , token , remove_from_cache = False ) <EOL> aff4 . FACTORY . Open ( sid , mode = "<STR_LIT:r>" , token = token ) <EOL> aff4 . FACTORY . Flush ( ) <EOL> self . RevokeClientApproval ( client_id , token , remove_from_cache = True ) <EOL> self . assertRaises ( access_control . UnauthorizedAccess , <EOL> aff4 . FACTORY . Open , sid , mode = "<STR_LIT:r>" , token = token ) <EOL> def testBreakGlass ( self ) : <EOL> """<STR_LIT>""" <EOL> client_id = rdf_client . ClientURN ( "<STR_LIT>" % <NUM_LIT:0> ) <EOL> urn = client_id . Add ( "<STR_LIT>" ) <EOL> self . assertRaises ( access_control . UnauthorizedAccess , aff4 . FACTORY . Open , urn , <EOL> token = self . token ) <EOL> email = { } <EOL> def SendEmail ( to , from_user , subject , message , ** _ ) : <EOL> email [ "<STR_LIT:to>" ] = to <EOL> email [ "<STR_LIT>" ] = from_user <EOL> email [ "<STR_LIT>" ] = subject <EOL> email [ "<STR_LIT:message>" ] = message <EOL> with utils . Stubber ( email_alerts . EMAIL_ALERTER , "<STR_LIT>" , SendEmail ) : <EOL> flow . GRRFlow . StartFlow ( <EOL> client_id = client_id , flow_name = "<STR_LIT>" , <EOL> token = self . token , reason = self . token . reason ) <EOL> self . token . is_emergency = False <EOL> aff4 . FACTORY . Open ( urn , token = self . token ) <EOL> self . assertEqual ( email [ "<STR_LIT:to>" ] , <EOL> config_lib . CONFIG [ "<STR_LIT>" ] ) <EOL> self . assertIn ( self . token . username , email [ "<STR_LIT:message>" ] ) <EOL> self . assertEqual ( email [ "<STR_LIT>" ] , self . token . username ) <EOL> self . assertEqual ( self . token . is_emergency , True ) <EOL> def testNonAdminsCanNotStartAdminOnlyFlow ( self ) : <EOL> noadmin_token = access_control . ACLToken ( username = "<STR_LIT>" , <EOL> reason = "<STR_LIT>" ) <EOL> with self . ACLChecksDisabled ( ) : <EOL> client_id = self . SetupClients ( <NUM_LIT:1> ) [ <NUM_LIT:0> ] <EOL> self . CreateUser ( "<STR_LIT>" ) <EOL> self . RequestAndGrantClientApproval ( client_id , token = noadmin_token ) <EOL> with self . assertRaises ( access_control . UnauthorizedAccess ) : <EOL> flow . GRRFlow . StartFlow ( flow_name = AdminOnlyFlow . __name__ , <EOL> client_id = client_id , token = noadmin_token , <EOL> sync = False ) <EOL> def testAdminsCanStartAdminOnlyFlow ( self ) : <EOL> admin_token = access_control . ACLToken ( username = "<STR_LIT>" , <EOL> reason = "<STR_LIT>" ) <EOL> with self . ACLChecksDisabled ( ) : <EOL> client_id = self . SetupClients ( <NUM_LIT:1> ) [ <NUM_LIT:0> ] <EOL> self . CreateAdminUser ( "<STR_LIT>" ) <EOL> self . RequestAndGrantClientApproval ( client_id , token = admin_token ) <EOL> flow . GRRFlow . StartFlow ( flow_name = AdminOnlyFlow . __name__ , <EOL> client_id = client_id , token = admin_token , <EOL> sync = False ) <EOL> def testNotAclEnforcedFlowCanBeStartedWithClient ( self ) : <EOL> with self . ACLChecksDisabled ( ) : <EOL> client_id = self . SetupClients ( <NUM_LIT:1> ) [ <NUM_LIT:0> ] <EOL> flow . GRRFlow . StartFlow ( flow_name = NotEnforcedFlow . __name__ , <EOL> client_id = client_id , token = self . token , sync = False ) <EOL> def testClientFlowWithoutCategoryCanNotBeStartedWithClient ( self ) : <EOL> with self . ACLChecksDisabled ( ) : <EOL> client_id = self . SetupClients ( <NUM_LIT:1> ) [ <NUM_LIT:0> ] <EOL> with self . assertRaises ( access_control . UnauthorizedAccess ) : <EOL> flow . GRRFlow . StartFlow ( flow_name = ClientFlowWithoutCategory . __name__ , <EOL> client_id = client_id , token = self . token ) <EOL> def testClientFlowWithCategoryCanBeStartedWithClient ( self ) : <EOL> with self . ACLChecksDisabled ( ) : <EOL> client_id = self . SetupClients ( <NUM_LIT:1> ) [ <NUM_LIT:0> ] <EOL> self . RequestAndGrantClientApproval ( client_id , self . token ) <EOL> flow . GRRFlow . StartFlow ( flow_name = ClientFlowWithCategory . __name__ , <EOL> client_id = client_id , token = self . token , sync = False ) <EOL> def testGlobalFlowWithoutCategoryCanNotBeStartedGlobally ( self ) : <EOL> with self . assertRaises ( access_control . UnauthorizedAccess ) : <EOL> flow . GRRFlow . StartFlow ( flow_name = GlobalFlowWithoutCategory . __name__ , <EOL> token = self . token , sync = False ) <EOL> def testGlobalFlowWithCategoryCanBeStartedGlobally ( self ) : <EOL> flow . GRRFlow . StartFlow ( flow_name = GlobalFlowWithCategory . __name__ , <EOL> token = self . token , sync = False ) <EOL> def testNotEnforcedFlowCanBeStartedGlobally ( self ) : <EOL> flow . GRRFlow . StartFlow ( flow_name = NotEnforcedFlow . __name__ , <EOL> token = self . token , sync = False ) <EOL> class ClientApprovalByLabelTests ( test_lib . GRRBaseTest ) : <EOL> """<STR_LIT>""" <EOL> install_mock_acl = False <EOL> def setUp ( self ) : <EOL> super ( ClientApprovalByLabelTests , self ) . setUp ( ) <EOL> client_ids = self . SetupClients ( <NUM_LIT:3> ) <EOL> self . client_nolabel = rdf_client . ClientURN ( client_ids [ <NUM_LIT:0> ] ) <EOL> self . client_legal = rdf_client . ClientURN ( client_ids [ <NUM_LIT:1> ] ) <EOL> self . client_prod = rdf_client . ClientURN ( client_ids [ <NUM_LIT:2> ] ) <EOL> with aff4 . FACTORY . Open ( self . client_legal , aff4_type = "<STR_LIT>" , <EOL> mode = "<STR_LIT>" , token = self . token ) as client_obj : <EOL> client_obj . AddLabels ( "<STR_LIT>" ) <EOL> with aff4 . FACTORY . Open ( self . client_prod , aff4_type = "<STR_LIT>" , <EOL> mode = "<STR_LIT>" , token = self . token ) as client_obj : <EOL> client_obj . AddLabels ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> self . db_manager_stubber = utils . Stubber ( <EOL> data_store . DB , "<STR_LIT>" , <EOL> access_control . FullAccessControlManager ( ) ) <EOL> self . db_manager_stubber . Start ( ) <EOL> self . approver = test_lib . ConfigOverrider ( <EOL> { "<STR_LIT>" : os . path . join ( <EOL> self . base_path , "<STR_LIT>" ) } ) <EOL> self . approver . Start ( ) <EOL> self . approval_manager_stubber = utils . Stubber ( <EOL> client_approval_auth , "<STR_LIT>" , <EOL> client_approval_auth . ClientApprovalAuthorizationManager ( ) ) <EOL> self . approval_manager_stubber . Start ( ) <EOL> def tearDown ( self ) : <EOL> self . db_manager_stubber . Stop ( ) <EOL> self . approval_manager_stubber . Stop ( ) <EOL> self . approver . Stop ( ) <EOL> def testClientNoLabels ( self ) : <EOL> nolabel_urn = self . client_nolabel . Add ( "<STR_LIT>" ) <EOL> token = access_control . ACLToken ( username = "<STR_LIT:test>" , reason = "<STR_LIT>" ) <EOL> self . assertRaises ( access_control . UnauthorizedAccess , aff4 . FACTORY . Open , <EOL> nolabel_urn , aff4_type = None , mode = "<STR_LIT>" , token = token ) <EOL> self . RequestAndGrantClientApproval ( self . client_nolabel , token ) <EOL> with aff4 . FACTORY . Open ( nolabel_urn , aff4_type = None , mode = "<STR_LIT>" , <EOL> token = token ) : <EOL> pass <EOL> def testClientApprovalSingleLabel ( self ) : <EOL> """<STR_LIT>""" <EOL> legal_urn = self . client_legal . Add ( "<STR_LIT>" ) <EOL> token = access_control . ACLToken ( username = "<STR_LIT:test>" , reason = "<STR_LIT>" ) <EOL> self . assertRaises ( access_control . UnauthorizedAccess , aff4 . FACTORY . Open , <EOL> legal_urn , aff4_type = None , mode = "<STR_LIT>" , token = token ) <EOL> self . RequestAndGrantClientApproval ( self . client_legal , token ) <EOL> self . assertRaises ( access_control . UnauthorizedAccess , aff4 . FACTORY . Open , <EOL> legal_urn , aff4_type = None , mode = "<STR_LIT>" , token = token ) <EOL> self . GrantClientApproval ( self . client_legal , token . username , <EOL> reason = token . reason , approver = "<STR_LIT>" ) <EOL> with aff4 . FACTORY . Open ( legal_urn , aff4_type = None , mode = "<STR_LIT>" , <EOL> token = token ) : <EOL> pass <EOL> def testClientApprovalMultiLabel ( self ) : <EOL> """<STR_LIT>""" <EOL> prod_urn = self . client_prod . Add ( "<STR_LIT>" ) <EOL> token = access_control . ACLToken ( username = "<STR_LIT>" , reason = "<STR_LIT>" ) <EOL> self . assertRaises ( access_control . UnauthorizedAccess , aff4 . FACTORY . Open , <EOL> prod_urn , aff4_type = None , mode = "<STR_LIT>" , token = token ) <EOL> self . RequestAndGrantClientApproval ( self . client_prod , token ) <EOL> self . assertRaises ( access_control . UnauthorizedAccess , aff4 . FACTORY . Open , <EOL> prod_urn , aff4_type = None , mode = "<STR_LIT>" , token = token ) <EOL> self . GrantClientApproval ( self . client_prod , token . username , <EOL> reason = token . reason , approver = "<STR_LIT>" ) <EOL> self . assertRaises ( access_control . UnauthorizedAccess , aff4 . FACTORY . Open , <EOL> prod_urn , aff4_type = None , mode = "<STR_LIT>" , token = token ) <EOL> self . GrantClientApproval ( self . client_prod , token . username , <EOL> reason = token . reason , approver = "<STR_LIT>" ) <EOL> self . assertRaises ( access_control . UnauthorizedAccess , aff4 . FACTORY . Open , <EOL> prod_urn , aff4_type = None , mode = "<STR_LIT>" , token = token ) <EOL> self . GrantClientApproval ( self . client_prod , token . username , <EOL> reason = token . reason , approver = "<STR_LIT>" ) <EOL> with aff4 . FACTORY . Open ( prod_urn , aff4_type = None , mode = "<STR_LIT>" , <EOL> token = token ) : <EOL> pass <EOL> def testClientApprovalMultiLabelCheckRequester ( self ) : <EOL> """<STR_LIT>""" <EOL> prod_urn = self . client_prod . Add ( "<STR_LIT>" ) <EOL> token = access_control . ACLToken ( username = "<STR_LIT>" , reason = "<STR_LIT>" ) <EOL> self . assertRaises ( access_control . UnauthorizedAccess , aff4 . FACTORY . Open , <EOL> prod_urn , aff4_type = None , mode = "<STR_LIT>" , token = token ) <EOL> self . RequestAndGrantClientApproval ( self . client_prod , token ) <EOL> self . GrantClientApproval ( self . client_prod , token . username , <EOL> reason = token . reason , approver = "<STR_LIT>" ) <EOL> self . GrantClientApproval ( self . client_prod , token . username , <EOL> reason = token . reason , approver = "<STR_LIT>" ) <EOL> self . GrantClientApproval ( self . client_prod , token . username , <EOL> reason = token . reason , approver = "<STR_LIT>" ) <EOL> self . assertRaises ( access_control . UnauthorizedAccess , aff4 . FACTORY . Open , <EOL> prod_urn , aff4_type = None , mode = "<STR_LIT>" , token = token ) <EOL> def main ( argv ) : <EOL> test_lib . GrrTestProgram ( argv = argv ) <EOL> if __name__ == "<STR_LIT:__main__>" : <EOL> flags . StartMain ( main ) </s>
