Unlock 30% off on Manual Testing Annual Plans this Holiday Season.

Copied to Clipboard!

How to use put_user_policy method in localstack

Best Python code snippet using localstack_python

test_iam.py

Source:test_iam.py

...18@attr(operation='Verify Put User Policy')19@attr(assertion='succeeds')20@attr('user-policy')21@attr('test_of_iam')22def test_put_user_policy():23    client = get_iam_client()24    policy_document = json.dumps(25        {"Version": "2012-10-17",26         "Statement": {27             "Effect": "Allow",28             "Action": "*",29             "Resource": "*"}}30    )31    response = client.put_user_policy(PolicyDocument=policy_document, PolicyName='AllAccessPolicy',32                                      UserName=get_alt_user_id())33    eq(response['ResponseMetadata']['HTTPStatusCode'], 200)34    response = client.delete_user_policy(PolicyName='AllAccessPolicy',35                                         UserName=get_alt_user_id())36    eq(response['ResponseMetadata']['HTTPStatusCode'], 200)37@attr(resource='user-policy')38@attr(method='put')39@attr(operation='Verify Put User Policy with invalid user')40@attr(assertion='succeeds')41@attr('user-policy')42@attr('test_of_iam')43def test_put_user_policy_invalid_user():44    client = get_iam_client()45    policy_document = json.dumps(46        {"Version": "2012-10-17",47         "Statement": {48             "Effect": "Allow",49             "Action": "*",50             "Resource": "*"}}51    )52    e = assert_raises(ClientError, client.put_user_policy, PolicyDocument=policy_document,53                      PolicyName='AllAccessPolicy', UserName="some-non-existing-user-id")54    status = _get_status(e.response)55    eq(status, 404)56@attr(resource='user-policy')57@attr(method='put')58@attr(operation='Verify Put User Policy using parameter value outside limit')59@attr(assertion='succeeds')60@attr('user-policy')61@attr('test_of_iam')62def test_put_user_policy_parameter_limit():63    client = get_iam_client()64    policy_document = json.dumps(65        {"Version": "2012-10-17",66         "Statement": [{67             "Effect": "Allow",68             "Action": "*",69             "Resource": "*"}] * 100070         }71    )72    e = assert_raises(ClientError, client.put_user_policy, PolicyDocument=policy_document,73                      PolicyName='AllAccessPolicy' * 10, UserName=get_alt_user_id())74    status = _get_status(e.response)75    eq(status, 400)76@attr(resource='user-policy')77@attr(method='put')78@attr(operation='Verify Put User Policy using invalid policy document elements')79@attr(assertion='succeeds')80@attr('user-policy')81@attr('test_of_iam')82@attr('fails_on_rgw')83def test_put_user_policy_invalid_element():84    client = get_iam_client()85    # With Version other than 2012-10-1786    policy_document = json.dumps(87        {"Version": "2010-10-17",88         "Statement": [{89             "Effect": "Allow",90             "Action": "*",91             "Resource": "*"}]92         }93    )94    e = assert_raises(ClientError, client.put_user_policy, PolicyDocument=policy_document,95                      PolicyName='AllAccessPolicy', UserName=get_alt_user_id())96    status = _get_status(e.response)97    eq(status, 400)98    # With no Statement99    policy_document = json.dumps(100        {101            "Version": "2012-10-17",102        }103    )104    e = assert_raises(ClientError, client.put_user_policy, PolicyDocument=policy_document,105                      PolicyName='AllAccessPolicy', UserName=get_alt_user_id())106    status = _get_status(e.response)107    eq(status, 400)108    # with same Sid for 2 statements109    policy_document = json.dumps(110        {"Version": "2012-10-17",111         "Statement": [112             {"Sid": "98AB54CF",113              "Effect": "Allow",114              "Action": "*",115              "Resource": "*"},116             {"Sid": "98AB54CF",117              "Effect": "Allow",118              "Action": "*",119              "Resource": "*"}]120         }121    )122    e = assert_raises(ClientError, client.put_user_policy, PolicyDocument=policy_document,123                      PolicyName='AllAccessPolicy', UserName=get_alt_user_id())124    status = _get_status(e.response)125    eq(status, 400)126    # with Principal127    policy_document = json.dumps(128        {"Version": "2012-10-17",129         "Statement": [{130             "Effect": "Allow",131             "Action": "*",132             "Resource": "*",133             "Principal": "arn:aws:iam:::username"}]134         }135    )136    e = assert_raises(ClientError, client.put_user_policy, PolicyDocument=policy_document,137                      PolicyName='AllAccessPolicy', UserName=get_alt_user_id())138    status = _get_status(e.response)139    eq(status, 400)140@attr(resource='user-policy')141@attr(method='put')142@attr(operation='Verify Put a policy that already exists')143@attr(assertion='succeeds')144@attr('user-policy')145@attr('test_of_iam')146def test_put_existing_user_policy():147    client = get_iam_client()148    policy_document = json.dumps(149        {"Version": "2012-10-17",150         "Statement": {151             "Effect": "Allow",152             "Action": "*",153             "Resource": "*"}154         }155    )156    response = client.put_user_policy(PolicyDocument=policy_document, PolicyName='AllAccessPolicy',157                                      UserName=get_alt_user_id())158    eq(response['ResponseMetadata']['HTTPStatusCode'], 200)159    client.put_user_policy(PolicyDocument=policy_document, PolicyName='AllAccessPolicy',160                           UserName=get_alt_user_id())161    client.delete_user_policy(PolicyName='AllAccessPolicy', UserName=get_alt_user_id())162@attr(resource='user-policy')163@attr(method='put')164@attr(operation='Verify List User policies')165@attr(assertion='succeeds')166@attr('user-policy')167@attr('test_of_iam')168def test_list_user_policy():169    client = get_iam_client()170    policy_document = json.dumps(171        {"Version": "2012-10-17",172         "Statement": {173             "Effect": "Allow",174             "Action": "*",175             "Resource": "*"}176         }177    )178    response = client.put_user_policy(PolicyDocument=policy_document, PolicyName='AllAccessPolicy',179                                      UserName=get_alt_user_id())180    eq(response['ResponseMetadata']['HTTPStatusCode'], 200)181    response = client.list_user_policies(UserName=get_alt_user_id())182    eq(response['ResponseMetadata']['HTTPStatusCode'], 200)183    client.delete_user_policy(PolicyName='AllAccessPolicy', UserName=get_alt_user_id())184@attr(resource='user-policy')185@attr(method='put')186@attr(operation='Verify List User policies with invalid user')187@attr(assertion='succeeds')188@attr('user-policy')189@attr('test_of_iam')190def test_list_user_policy_invalid_user():191    client = get_iam_client()192    e = assert_raises(ClientError, client.list_user_policies, UserName="some-non-existing-user-id")193    status = _get_status(e.response)194    eq(status, 404)195@attr(resource='user-policy')196@attr(method='get')197@attr(operation='Verify Get User policy')198@attr(assertion='succeeds')199@attr('user-policy')200@attr('test_of_iam')201def test_get_user_policy():202    client = get_iam_client()203    policy_document = json.dumps(204        {"Version": "2012-10-17",205         "Statement": {206             "Effect": "Allow",207             "Action": "*",208             "Resource": "*"}}209    )210    response = client.put_user_policy(PolicyDocument=policy_document, PolicyName='AllAccessPolicy',211                                      UserName=get_alt_user_id())212    eq(response['ResponseMetadata']['HTTPStatusCode'], 200)213    response = client.get_user_policy(PolicyName='AllAccessPolicy', UserName=get_alt_user_id())214    eq(response['ResponseMetadata']['HTTPStatusCode'], 200)215    response = client.delete_user_policy(PolicyName='AllAccessPolicy',216                                         UserName=get_alt_user_id())217    eq(response['ResponseMetadata']['HTTPStatusCode'], 200)218@attr(resource='user-policy')219@attr(method='get')220@attr(operation='Verify Get User Policy with invalid user')221@attr(assertion='succeeds')222@attr('user-policy')223@attr('test_of_iam')224def test_get_user_policy_invalid_user():225    client = get_iam_client()226    policy_document = json.dumps(227        {"Version": "2012-10-17",228         "Statement": {229             "Effect": "Allow",230             "Action": "*",231             "Resource": "*"}}232    )233    response = client.put_user_policy(PolicyDocument=policy_document, PolicyName='AllAccessPolicy',234                                      UserName=get_alt_user_id())235    eq(response['ResponseMetadata']['HTTPStatusCode'], 200)236    e = assert_raises(ClientError, client.get_user_policy, PolicyName='AllAccessPolicy',237                      UserName="some-non-existing-user-id")238    status = _get_status(e.response)239    eq(status, 404)240    client.delete_user_policy(PolicyName='AllAccessPolicy', UserName=get_alt_user_id())241@attr(resource='user-policy')242@attr(method='get')243@attr(operation='Verify Get User Policy with invalid policy name')244@attr(assertion='succeeds')245@attr('user-policy')246@attr('test_of_iam')247@attr('fails_on_rgw')248def test_get_user_policy_invalid_policy_name():249    client = get_iam_client()250    policy_document = json.dumps(251        {"Version": "2012-10-17",252         "Statement": {253             "Effect": "Allow",254             "Action": "*",255             "Resource": "*"}}256    )257    client.put_user_policy(PolicyDocument=policy_document, PolicyName='AllAccessPolicy',258                           UserName=get_alt_user_id())259    e = assert_raises(ClientError, client.get_user_policy, PolicyName='non-existing-policy-name',260                      UserName=get_alt_user_id())261    status = _get_status(e.response)262    eq(status, 404)263    client.delete_user_policy(PolicyName='AllAccessPolicy', UserName=get_alt_user_id())264@attr(resource='user-policy')265@attr(method='get')266@attr(operation='Verify Get Deleted User Policy')267@attr(assertion='succeeds')268@attr('user-policy')269@attr('test_of_iam')270@attr('fails_on_rgw')271def test_get_deleted_user_policy():272    client = get_iam_client()273    policy_document = json.dumps(274        {"Version": "2012-10-17",275         "Statement": {276             "Effect": "Allow",277             "Action": "*",278             "Resource": "*"}}279    )280    client.put_user_policy(PolicyDocument=policy_document, PolicyName='AllAccessPolicy',281                           UserName=get_alt_user_id())282    client.delete_user_policy(PolicyName='AllAccessPolicy', UserName=get_alt_user_id())283    e = assert_raises(ClientError, client.get_user_policy, PolicyName='AllAccessPolicy',284                      UserName=get_alt_user_id())285    status = _get_status(e.response)286    eq(status, 404)287@attr(resource='user-policy')288@attr(method='get')289@attr(operation='Verify Get a policy from multiple policies for a user')290@attr(assertion='succeeds')291@attr('user-policy')292@attr('test_of_iam')293def test_get_user_policy_from_multiple_policies():294    client = get_iam_client()295    policy_document_allow = json.dumps(296        {"Version": "2012-10-17",297         "Statement": {298             "Effect": "Allow",299             "Action": "*",300             "Resource": "*"}}301    )302    response = client.put_user_policy(PolicyDocument=policy_document_allow,303                                      PolicyName='AllowAccessPolicy1',304                                      UserName=get_alt_user_id())305    eq(response['ResponseMetadata']['HTTPStatusCode'], 200)306    response = client.put_user_policy(PolicyDocument=policy_document_allow,307                                      PolicyName='AllowAccessPolicy2',308                                      UserName=get_alt_user_id())309    eq(response['ResponseMetadata']['HTTPStatusCode'], 200)310    response = client.get_user_policy(PolicyName='AllowAccessPolicy2',311                                      UserName=get_alt_user_id())312    eq(response['ResponseMetadata']['HTTPStatusCode'], 200)313    response = client.delete_user_policy(PolicyName='AllowAccessPolicy1',314                                         UserName=get_alt_user_id())315    eq(response['ResponseMetadata']['HTTPStatusCode'], 200)316    response = client.delete_user_policy(PolicyName='AllowAccessPolicy2',317                                         UserName=get_alt_user_id())318    eq(response['ResponseMetadata']['HTTPStatusCode'], 200)319@attr(resource='user-policy')320@attr(method='delete')321@attr(operation='Verify Delete User Policy')322@attr(assertion='succeeds')323@attr('user-policy')324@attr('test_of_iam')325def test_delete_user_policy():326    client = get_iam_client()327    policy_document_allow = json.dumps(328        {"Version": "2012-10-17",329         "Statement": {330             "Effect": "Allow",331             "Action": "*",332             "Resource": "*"}}333    )334    response = client.put_user_policy(PolicyDocument=policy_document_allow,335                                      PolicyName='AllowAccessPolicy',336                                      UserName=get_alt_user_id())337    eq(response['ResponseMetadata']['HTTPStatusCode'], 200)338    response = client.delete_user_policy(PolicyName='AllowAccessPolicy',339                                         UserName=get_alt_user_id())340    eq(response['ResponseMetadata']['HTTPStatusCode'], 200)341@attr(resource='user-policy')342@attr(method='delete')343@attr(operation='Verify Delete User Policy with invalid user')344@attr(assertion='succeeds')345@attr('user-policy')346@attr('test_of_iam')347def test_delete_user_policy_invalid_user():348    client = get_iam_client()349    policy_document_allow = json.dumps(350        {"Version": "2012-10-17",351         "Statement": {352             "Effect": "Allow",353             "Action": "*",354             "Resource": "*"}}355    )356    response = client.put_user_policy(PolicyDocument=policy_document_allow,357                                      PolicyName='AllowAccessPolicy',358                                      UserName=get_alt_user_id())359    eq(response['ResponseMetadata']['HTTPStatusCode'], 200)360    e = assert_raises(ClientError, client.delete_user_policy, PolicyName='AllAccessPolicy',361                      UserName="some-non-existing-user-id")362    status = _get_status(e.response)363    eq(status, 404)364    response = client.delete_user_policy(PolicyName='AllowAccessPolicy',365                                         UserName=get_alt_user_id())366    eq(response['ResponseMetadata']['HTTPStatusCode'], 200)367@attr(resource='user-policy')368@attr(method='delete')369@attr(operation='Verify Delete User Policy with invalid policy name')370@attr(assertion='succeeds')371@attr('user-policy')372@attr('test_of_iam')373def test_delete_user_policy_invalid_policy_name():374    client = get_iam_client()375    policy_document_allow = json.dumps(376        {"Version": "2012-10-17",377         "Statement": {378             "Effect": "Allow",379             "Action": "*",380             "Resource": "*"}}381    )382    response = client.put_user_policy(PolicyDocument=policy_document_allow,383                                      PolicyName='AllowAccessPolicy',384                                      UserName=get_alt_user_id())385    eq(response['ResponseMetadata']['HTTPStatusCode'], 200)386    e = assert_raises(ClientError, client.delete_user_policy, PolicyName='non-existing-policy-name',387                      UserName=get_alt_user_id())388    status = _get_status(e.response)389    eq(status, 404)390    response = client.delete_user_policy(PolicyName='AllowAccessPolicy',391                                         UserName=get_alt_user_id())392    eq(response['ResponseMetadata']['HTTPStatusCode'], 200)393@attr(resource='user-policy')394@attr(method='delete')395@attr(operation='Verify Delete multiple User policies for a user')396@attr(assertion='succeeds')397@attr('user-policy')398@attr('test_of_iam')399def test_delete_user_policy_from_multiple_policies():400    client = get_iam_client()401    policy_document_allow = json.dumps(402        {"Version": "2012-10-17",403         "Statement": {404             "Effect": "Allow",405             "Action": "*",406             "Resource": "*"}}407    )408    response = client.put_user_policy(PolicyDocument=policy_document_allow,409                                      PolicyName='AllowAccessPolicy1',410                                      UserName=get_alt_user_id())411    eq(response['ResponseMetadata']['HTTPStatusCode'], 200)412    response = client.put_user_policy(PolicyDocument=policy_document_allow,413                                      PolicyName='AllowAccessPolicy2',414                                      UserName=get_alt_user_id())415    eq(response['ResponseMetadata']['HTTPStatusCode'], 200)416    response = client.put_user_policy(PolicyDocument=policy_document_allow,417                                      PolicyName='AllowAccessPolicy3',418                                      UserName=get_alt_user_id())419    eq(response['ResponseMetadata']['HTTPStatusCode'], 200)420    response = client.delete_user_policy(PolicyName='AllowAccessPolicy1',421                                         UserName=get_alt_user_id())422    eq(response['ResponseMetadata']['HTTPStatusCode'], 200)423    response = client.delete_user_policy(PolicyName='AllowAccessPolicy2',424                                         UserName=get_alt_user_id())425    eq(response['ResponseMetadata']['HTTPStatusCode'], 200)426    response = client.get_user_policy(PolicyName='AllowAccessPolicy3',427                                      UserName=get_alt_user_id())428    eq(response['ResponseMetadata']['HTTPStatusCode'], 200)429    response = client.delete_user_policy(PolicyName='AllowAccessPolicy3',430                                         UserName=get_alt_user_id())431    eq(response['ResponseMetadata']['HTTPStatusCode'], 200)432@attr(resource='user-policy')433@attr(method='s3 Actions')434@attr(operation='Verify Allow Bucket Actions in user Policy')435@attr(assertion='succeeds')436@attr('user-policy')437@attr('test_of_iam')438def test_allow_bucket_actions_in_user_policy():439    client = get_iam_client()440    s3_client_alt = get_alt_client()441    s3_client_iam = get_iam_s3client()442    bucket = get_new_bucket(client=s3_client_iam)443    s3_client_iam.put_object(Bucket=bucket, Key='foo', Body='bar')444    policy_document_allow = json.dumps(445        {"Version": "2012-10-17",446         "Statement": {447             "Effect": "Allow",448             "Action": ["s3:ListBucket", "s3:DeleteBucket"],449             "Resource": f"arn:aws:s3:::{bucket}"}}450    )451    response = client.put_user_policy(PolicyDocument=policy_document_allow,452                                      PolicyName='AllowAccessPolicy', UserName=get_alt_user_id())453    eq(response['ResponseMetadata']['HTTPStatusCode'], 200)454    response = s3_client_alt.list_objects(Bucket=bucket)455    object_found = False456    for object_received in response['Contents']:457        if "foo" == object_received['Key']:458            object_found = True459            break460    if not object_found:461        raise AssertionError("Object is not listed")462    response = s3_client_iam.delete_object(Bucket=bucket, Key='foo')463    eq(response['ResponseMetadata']['HTTPStatusCode'], 204)464    response = s3_client_alt.delete_bucket(Bucket=bucket)465    eq(response['ResponseMetadata']['HTTPStatusCode'], 204)466    response = s3_client_iam.list_buckets()467    for bucket in response['Buckets']:468        if bucket == bucket['Name']:469            raise AssertionError("deleted bucket is getting listed")470    response = client.delete_user_policy(PolicyName='AllowAccessPolicy',471                                         UserName=get_alt_user_id())472    eq(response['ResponseMetadata']['HTTPStatusCode'], 200)473@attr(resource='user-policy')474@attr(method='s3 Actions')475@attr(operation='Verify Deny Bucket Actions in user Policy')476@attr(assertion='succeeds')477@attr('user-policy')478@attr('test_of_iam')479@attr('fails_on_dbstore')480def test_deny_bucket_actions_in_user_policy():481    client = get_iam_client()482    s3_client = get_alt_client()483    bucket = get_new_bucket(client=s3_client)484    policy_document_deny = json.dumps(485        {"Version": "2012-10-17",486         "Statement": {487             "Effect": "Deny",488             "Action": ["s3:ListAllMyBuckets", "s3:DeleteBucket"],489             "Resource": "arn:aws:s3:::*"}}490    )491    response = client.put_user_policy(PolicyDocument=policy_document_deny,492                                      PolicyName='DenyAccessPolicy',493                                      UserName=get_alt_user_id())494    eq(response['ResponseMetadata']['HTTPStatusCode'], 200)495    e = assert_raises(ClientError, s3_client.list_buckets, Bucket=bucket)496    status, error_code = _get_status_and_error_code(e.response)497    eq(status, 403)498    eq(error_code, 'AccessDenied')499    e = assert_raises(ClientError, s3_client.delete_bucket, Bucket=bucket)500    status, error_code = _get_status_and_error_code(e.response)501    eq(status, 403)502    eq(error_code, 'AccessDenied')503    response = client.delete_user_policy(PolicyName='DenyAccessPolicy',504                                         UserName=get_alt_user_id())505    eq(response['ResponseMetadata']['HTTPStatusCode'], 200)506    response = s3_client.delete_bucket(Bucket=bucket)507    eq(response['ResponseMetadata']['HTTPStatusCode'], 204)508@attr(resource='user-policy')509@attr(method='s3 Actions')510@attr(operation='Verify Allow Object Actions in user Policy')511@attr(assertion='succeeds')512@attr('user-policy')513@attr('test_of_iam')514def test_allow_object_actions_in_user_policy():515    client = get_iam_client()516    s3_client_alt = get_alt_client()517    s3_client_iam = get_iam_s3client()518    bucket = get_new_bucket(client=s3_client_iam)519    policy_document_allow = json.dumps(520        {"Version": "2012-10-17",521         "Statement": {522             "Effect": "Allow",523             "Action": ["s3:PutObject", "s3:GetObject", "s3:DeleteObject"],524             "Resource": f"arn:aws:s3:::{bucket}/*"}}525    )526    response = client.put_user_policy(PolicyDocument=policy_document_allow,527                                      PolicyName='AllowAccessPolicy', UserName=get_alt_user_id())528    eq(response['ResponseMetadata']['HTTPStatusCode'], 200)529    s3_client_alt.put_object(Bucket=bucket, Key='foo', Body='bar')530    response = s3_client_alt.get_object(Bucket=bucket, Key='foo')531    body = response['Body'].read()532    if type(body) is bytes:533        body = body.decode()534    eq(body, "bar")535    response = s3_client_alt.delete_object(Bucket=bucket, Key='foo')536    eq(response['ResponseMetadata']['HTTPStatusCode'], 204)537    e = assert_raises(ClientError, s3_client_iam.get_object, Bucket=bucket, Key='foo')538    status, error_code = _get_status_and_error_code(e.response)539    eq(status, 404)540    eq(error_code, 'NoSuchKey')541    response = s3_client_iam.delete_bucket(Bucket=bucket)542    eq(response['ResponseMetadata']['HTTPStatusCode'], 204)543    response = client.delete_user_policy(PolicyName='AllowAccessPolicy',544                                         UserName=get_alt_user_id())545    eq(response['ResponseMetadata']['HTTPStatusCode'], 200)546@attr(resource='user-policy')547@attr(method='s3 Actions')548@attr(operation='Verify Deny Object Actions in user Policy')549@attr(assertion='succeeds')550@attr('user-policy')551@attr('test_of_iam')552@attr('fails_on_dbstore')553def test_deny_object_actions_in_user_policy():554    client = get_iam_client()555    s3_client_alt = get_alt_client()556    bucket = get_new_bucket(client=s3_client_alt)557    s3_client_alt.put_object(Bucket=bucket, Key='foo', Body='bar')558    policy_document_deny = json.dumps(559        {"Version": "2012-10-17",560         "Statement": [{561             "Effect": "Deny",562             "Action": ["s3:PutObject", "s3:GetObject", "s3:DeleteObject"],563             "Resource": f"arn:aws:s3:::{bucket}/*"}, {564             "Effect": "Allow",565             "Action": ["s3:DeleteBucket"],566             "Resource": f"arn:aws:s3:::{bucket}"}]}567    )568    client.put_user_policy(PolicyDocument=policy_document_deny, PolicyName='DenyAccessPolicy',569                           UserName=get_alt_user_id())570    e = assert_raises(ClientError, s3_client_alt.put_object, Bucket=bucket, Key='foo')571    status, error_code = _get_status_and_error_code(e.response)572    eq(status, 403)573    eq(error_code, 'AccessDenied')574    e = assert_raises(ClientError, s3_client_alt.get_object, Bucket=bucket, Key='foo')575    status, error_code = _get_status_and_error_code(e.response)576    eq(status, 403)577    eq(error_code, 'AccessDenied')578    e = assert_raises(ClientError, s3_client_alt.delete_object, Bucket=bucket, Key='foo')579    status, error_code = _get_status_and_error_code(e.response)580    eq(status, 403)581    eq(error_code, 'AccessDenied')582    response = client.delete_user_policy(PolicyName='DenyAccessPolicy',583                                         UserName=get_alt_user_id())584    eq(response['ResponseMetadata']['HTTPStatusCode'], 200)585@attr(resource='user-policy')586@attr(method='s3 Actions')587@attr(operation='Verify Allow Multipart Actions in user Policy')588@attr(assertion='succeeds')589@attr('user-policy')590@attr('test_of_iam')591def test_allow_multipart_actions_in_user_policy():592    client = get_iam_client()593    s3_client_alt = get_alt_client()594    s3_client_iam = get_iam_s3client()595    bucket = get_new_bucket(client=s3_client_iam)596    policy_document_allow = json.dumps(597        {"Version": "2012-10-17",598         "Statement": {599             "Effect": "Allow",600             "Action": ["s3:ListBucketMultipartUploads", "s3:AbortMultipartUpload"],601             "Resource": "arn:aws:s3:::*"}}602    )603    response = client.put_user_policy(PolicyDocument=policy_document_allow,604                                      PolicyName='AllowAccessPolicy', UserName=get_alt_user_id())605    eq(response['ResponseMetadata']['HTTPStatusCode'], 200)606    key = "mymultipart"607    mb = 1024 * 1024608    (upload_id, _, _) = _multipart_upload(client=s3_client_iam, bucket_name=bucket, key=key,609                                          size=5 * mb)610    response = s3_client_alt.list_multipart_uploads(Bucket=bucket)611    eq(response['ResponseMetadata']['HTTPStatusCode'], 200)612    response = s3_client_alt.abort_multipart_upload(Bucket=bucket, Key=key, UploadId=upload_id)613    eq(response['ResponseMetadata']['HTTPStatusCode'], 204)614    response = s3_client_iam.delete_bucket(Bucket=bucket)615    eq(response['ResponseMetadata']['HTTPStatusCode'], 204)616    response = client.delete_user_policy(PolicyName='AllowAccessPolicy',617                                         UserName=get_alt_user_id())618    eq(response['ResponseMetadata']['HTTPStatusCode'], 200)619@attr(resource='user-policy')620@attr(method='s3 Actions')621@attr(operation='Verify Deny Multipart Actions in user Policy')622@attr(assertion='succeeds')623@attr('user-policy')624@attr('test_of_iam')625@attr('fails_on_dbstore')626def test_deny_multipart_actions_in_user_policy():627    client = get_iam_client()628    s3_client = get_alt_client()629    bucket = get_new_bucket(client=s3_client)630    policy_document_deny = json.dumps(631        {"Version": "2012-10-17",632         "Statement": {633             "Effect": "Deny",634             "Action": ["s3:ListBucketMultipartUploads", "s3:AbortMultipartUpload"],635             "Resource": "arn:aws:s3:::*"}}636    )637    response = client.put_user_policy(PolicyDocument=policy_document_deny,638                                      PolicyName='DenyAccessPolicy',639                                      UserName=get_alt_user_id())640    eq(response['ResponseMetadata']['HTTPStatusCode'], 200)641    key = "mymultipart"642    mb = 1024 * 1024643    (upload_id, _, _) = _multipart_upload(client=s3_client, bucket_name=bucket, key=key,644                                          size=5 * mb)645    e = assert_raises(ClientError, s3_client.list_multipart_uploads, Bucket=bucket)646    status, error_code = _get_status_and_error_code(e.response)647    eq(status, 403)648    eq(error_code, 'AccessDenied')649    e = assert_raises(ClientError, s3_client.abort_multipart_upload, Bucket=bucket,650                      Key=key, UploadId=upload_id)651    status, error_code = _get_status_and_error_code(e.response)652    eq(status, 403)653    eq(error_code, 'AccessDenied')654    response = s3_client.delete_bucket(Bucket=bucket)655    eq(response['ResponseMetadata']['HTTPStatusCode'], 204)656    response = client.delete_user_policy(PolicyName='DenyAccessPolicy',657                                         UserName=get_alt_user_id())658    eq(response['ResponseMetadata']['HTTPStatusCode'], 200)659@attr(resource='user-policy')660@attr(method='s3 Actions')661@attr(operation='Verify Allow Tagging Actions in user Policy')662@attr(assertion='succeeds')663@attr('user-policy')664@attr('test_of_iam')665@attr('fails_on_dbstore')666def test_allow_tagging_actions_in_user_policy():667    client = get_iam_client()668    s3_client_alt = get_alt_client()669    s3_client_iam = get_iam_s3client()670    bucket = get_new_bucket(client=s3_client_iam)671    policy_document_allow = json.dumps(672        {"Version": "2012-10-17",673         "Statement": {674             "Effect": "Allow",675             "Action": ["s3:PutBucketTagging", "s3:GetBucketTagging",676                        "s3:PutObjectTagging", "s3:GetObjectTagging"],677             "Resource": f"arn:aws:s3:::*"}}678    )679    client.put_user_policy(PolicyDocument=policy_document_allow, PolicyName='AllowAccessPolicy',680                           UserName=get_alt_user_id())681    tags = {'TagSet': [{'Key': 'Hello', 'Value': 'World'}, ]}682    response = s3_client_alt.put_bucket_tagging(Bucket=bucket, Tagging=tags)683    eq(response['ResponseMetadata']['HTTPStatusCode'], 200)684    response = s3_client_alt.get_bucket_tagging(Bucket=bucket)685    eq(response['ResponseMetadata']['HTTPStatusCode'], 200)686    eq(response['TagSet'][0]['Key'], 'Hello')687    eq(response['TagSet'][0]['Value'], 'World')688    obj_key = 'obj'689    response = s3_client_iam.put_object(Bucket=bucket, Key=obj_key, Body='obj_body')690    eq(response['ResponseMetadata']['HTTPStatusCode'], 200)691    response = s3_client_alt.put_object_tagging(Bucket=bucket, Key=obj_key, Tagging=tags)692    eq(response['ResponseMetadata']['HTTPStatusCode'], 200)693    response = s3_client_alt.get_object_tagging(Bucket=bucket, Key=obj_key)694    eq(response['ResponseMetadata']['HTTPStatusCode'], 200)695    eq(response['TagSet'], tags['TagSet'])696    response = s3_client_iam.delete_object(Bucket=bucket, Key=obj_key)697    eq(response['ResponseMetadata']['HTTPStatusCode'], 204)698    response = s3_client_iam.delete_bucket(Bucket=bucket)699    eq(response['ResponseMetadata']['HTTPStatusCode'], 204)700    response = client.delete_user_policy(PolicyName='AllowAccessPolicy',701                                         UserName=get_alt_user_id())702    eq(response['ResponseMetadata']['HTTPStatusCode'], 200)703@attr(resource='user-policy')704@attr(method='s3 Actions')705@attr(operation='Verify Deny Tagging Actions in user Policy')706@attr(assertion='succeeds')707@attr('user-policy')708@attr('test_of_iam')709@attr('fails_on_dbstore')710def test_deny_tagging_actions_in_user_policy():711    client = get_iam_client()712    s3_client = get_alt_client()713    bucket = get_new_bucket(client=s3_client)714    policy_document_deny = json.dumps(715        {"Version": "2012-10-17",716         "Statement": {717             "Effect": "Deny",718             "Action": ["s3:PutBucketTagging", "s3:GetBucketTagging",719                        "s3:PutObjectTagging", "s3:DeleteObjectTagging"],720             "Resource": "arn:aws:s3:::*"}}721    )722    client.put_user_policy(PolicyDocument=policy_document_deny, PolicyName='DenyAccessPolicy',723                           UserName=get_alt_user_id())724    tags = {'TagSet': [{'Key': 'Hello', 'Value': 'World'}, ]}725    e = assert_raises(ClientError, s3_client.put_bucket_tagging, Bucket=bucket, Tagging=tags)726    status, error_code = _get_status_and_error_code(e.response)727    eq(status, 403)728    eq(error_code, 'AccessDenied')729    e = assert_raises(ClientError, s3_client.get_bucket_tagging, Bucket=bucket)730    status, error_code = _get_status_and_error_code(e.response)731    eq(status, 403)732    eq(error_code, 'AccessDenied')733    obj_key = 'obj'734    response = s3_client.put_object(Bucket=bucket, Key=obj_key, Body='obj_body')735    eq(response['ResponseMetadata']['HTTPStatusCode'], 200)736    e = assert_raises(ClientError, s3_client.put_object_tagging, Bucket=bucket, Key=obj_key,737                      Tagging=tags)738    status, error_code = _get_status_and_error_code(e.response)739    eq(status, 403)740    eq(error_code, 'AccessDenied')741    e = assert_raises(ClientError, s3_client.delete_object_tagging, Bucket=bucket, Key=obj_key)742    status, error_code = _get_status_and_error_code(e.response)743    eq(status, 403)744    eq(error_code, 'AccessDenied')745    response = s3_client.delete_object(Bucket=bucket, Key=obj_key)746    eq(response['ResponseMetadata']['HTTPStatusCode'], 204)747    response = s3_client.delete_bucket(Bucket=bucket)748    eq(response['ResponseMetadata']['HTTPStatusCode'], 204)749    response = client.delete_user_policy(PolicyName='DenyAccessPolicy',750                                         UserName=get_alt_user_id())751    eq(response['ResponseMetadata']['HTTPStatusCode'], 200)752@attr(resource='user-policy')753@attr(method='put')754@attr(operation='Verify conflicting user policy statements')755@attr(assertion='succeeds')756@attr('user-policy')757@attr('test_of_iam')758@attr('fails_on_dbstore')759def test_verify_conflicting_user_policy_statements():760    s3client = get_alt_client()761    bucket = get_new_bucket(client=s3client)762    policy_document = json.dumps(763        {"Version": "2012-10-17",764         "Statement": [765             {"Sid": "98AB54CG",766              "Effect": "Allow",767              "Action": "s3:ListBucket",768              "Resource": f"arn:aws:s3:::{bucket}"},769             {"Sid": "98AB54CA",770              "Effect": "Deny",771              "Action": "s3:ListBucket",772              "Resource": f"arn:aws:s3:::{bucket}"}773         ]}774    )775    client = get_iam_client()776    response = client.put_user_policy(PolicyDocument=policy_document, PolicyName='DenyAccessPolicy',777                                      UserName=get_alt_user_id())778    eq(response['ResponseMetadata']['HTTPStatusCode'], 200)779    e = assert_raises(ClientError, s3client.list_objects, Bucket=bucket)780    status, error_code = _get_status_and_error_code(e.response)781    eq(status, 403)782    eq(error_code, 'AccessDenied')783    response = client.delete_user_policy(PolicyName='DenyAccessPolicy',784                                         UserName=get_alt_user_id())785    eq(response['ResponseMetadata']['HTTPStatusCode'], 200)786@attr(resource='user-policy')787@attr(method='put')788@attr(operation='Verify conflicting user policies')789@attr(assertion='succeeds')790@attr('user-policy')791@attr('test_of_iam')792@attr('fails_on_dbstore')793def test_verify_conflicting_user_policies():794    s3client = get_alt_client()795    bucket = get_new_bucket(client=s3client)796    policy_allow = json.dumps(797        {"Version": "2012-10-17",798         "Statement": {"Sid": "98AB54CG",799                       "Effect": "Allow",800                       "Action": "s3:ListBucket",801                       "Resource": f"arn:aws:s3:::{bucket}"}}802    )803    policy_deny = json.dumps(804        {"Version": "2012-10-17",805         "Statement": {"Sid": "98AB54CGZ",806                       "Effect": "Deny",807                       "Action": "s3:ListBucket",808                       "Resource": f"arn:aws:s3:::{bucket}"}}809    )810    client = get_iam_client()811    response = client.put_user_policy(PolicyDocument=policy_allow, PolicyName='AllowAccessPolicy',812                                      UserName=get_alt_user_id())813    eq(response['ResponseMetadata']['HTTPStatusCode'], 200)814    response = client.put_user_policy(PolicyDocument=policy_deny, PolicyName='DenyAccessPolicy',815                                      UserName=get_alt_user_id())816    eq(response['ResponseMetadata']['HTTPStatusCode'], 200)817    e = assert_raises(ClientError, s3client.list_objects, Bucket=bucket)818    status, error_code = _get_status_and_error_code(e.response)819    eq(status, 403)820    eq(error_code, 'AccessDenied')821    response = client.delete_user_policy(PolicyName='AllowAccessPolicy',822                                         UserName=get_alt_user_id())823    eq(response['ResponseMetadata']['HTTPStatusCode'], 200)824    response = client.delete_user_policy(PolicyName='DenyAccessPolicy',825                                         UserName=get_alt_user_id())826    eq(response['ResponseMetadata']['HTTPStatusCode'], 200)827@attr(resource='user-policy')828@attr(operation='Verify Allow Actions for IAM user policies')829@attr(assertion='succeeds')830@attr('user-policy')831@attr('test_of_iam')832def test_verify_allow_iam_actions():833    policy1 = json.dumps(834        {"Version": "2012-10-17",835         "Statement": {"Sid": "98AB54CGA",836                       "Effect": "Allow",837                       "Action": ["iam:PutUserPolicy", "iam:GetUserPolicy",838                                  "iam:ListUserPolicies", "iam:DeleteUserPolicy"],839                       "Resource": f"arn:aws:iam:::user/{get_alt_user_id()}"}}840    )841    client1 = get_iam_client()842    iam_client_alt = get_alt_iam_client()843    response = client1.put_user_policy(PolicyDocument=policy1, PolicyName='AllowAccessPolicy',844                                       UserName=get_alt_user_id())845    eq(response['ResponseMetadata']['HTTPStatusCode'], 200)846    response = iam_client_alt.get_user_policy(PolicyName='AllowAccessPolicy',847                                       UserName=get_alt_user_id())848    eq(response['ResponseMetadata']['HTTPStatusCode'], 200)849    response = iam_client_alt.list_user_policies(UserName=get_alt_user_id())850    eq(response['ResponseMetadata']['HTTPStatusCode'], 200)851    response = iam_client_alt.delete_user_policy(PolicyName='AllowAccessPolicy',852                                          UserName=get_alt_user_id())...

awsmanager.py

Source:awsmanager.py

...18    def delete_user(self, username=None):19        return self.aws_manager.IAM.delete_user(username=username)20    def get_user(self, username=None):21        return self.aws_manager.IAM.get_user(username=username)22    def put_user_policy(self, username=None, policyname=None, policydocument=None):23        return self.aws_manager.IAM.put_user_policy(username=username, policyname=policyname,24            policydocument=policydocument)25    def get_usernames(self):26        return self.aws_manager.IAM.get_usernames()27    def create_user_key_pair(self, username=None):28        return self.aws_manager.IAM.create_user_key_pair(username=username)29    def get_s3_policy(self, bucket=None, folder=None):30        return self.aws_manager.S3.get_s3_policy(bucket=bucket, folder=folder)31    def get_sqs_policy(self, queue_name=None, full_access=None):32        return self.aws_manager.SQS.get_sqs_policy(queue_name=queue_name, full_access=full_access)33    def allow_sqs_queue_for_user(self, username=None, queue_name=None, full_access=None):34        policy = self.get_sqs_policy(queue_name=queue_name, full_access=full_access)35        self.put_user_policy(username=username, policyname='sqs_policy_%s_%s'%(username, queue_name),36            policydocument=policy)37    def create_s3_for_user(self, username=None, bucket=None, region=None):38        return self.aws_manager.S3.create_s3_for_user(username=username,39            bucket=bucket, region=region)40    def allow_s3_folder_for_user(self, username=None, bucket=None, folder=None):41        return self.aws_manager.S3.allow_s3_folder_for_user(username=username,42            bucket=bucket, folder=folder)43    def get_hosted_zones(self):44        return self.aws_manager.Route53.get_hosted_zones()45class ServiceParameters(BaseComponent):46    def service_parameters(self,pane,datapath=None,**kwargs):47        fb = pane.formbuilder(datapath=datapath)48        fb.textbox(value='^.aws_access_key_id',lbl='Aws Access Key Id')49        fb.textbox(value='^.aws_secret_access_key',lbl='Aws Secret Access Key')...

create_config_aws_event_relay_user.py

Source:create_config_aws_event_relay_user.py

...22        UserName=username23    )24except IAM.exceptions.EntityAlreadyExistsException:25    pass26IAM.put_user_policy(27    UserName=username,28    PolicyName='sns_publisher',29    PolicyDocument=json.dumps({30        'Version': '2012-10-17',31        'Statement': [32            {33                'Action': [34                    'sns:Publish'35                ],36                'Effect': 'Allow',37                'Resource': sns_arn38            }39        ]40    })41)42IAM.put_user_policy(43    UserName=username,44    PolicyName='sqs_sender',45    PolicyDocument=json.dumps({46        'Version': '2012-10-17',47        'Statement': [48            {49                'Action': [50                    'sqs:SendMessage'51                ],52                'Effect': 'Allow',53                'Resource': sqs_arn54            }55        ]56    })...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.