(c: Composition, ctx: TestContext)
| 122 | |
| 123 | |
| 124 | def test_credentials(c: Composition, ctx: TestContext): |
| 125 | # Create a user with an access key. |
| 126 | customer_user = f"testdrive-{ctx.seed}-Customer" |
| 127 | ctx.iam.create_user(UserName=customer_user) |
| 128 | access_key = ctx.iam.create_access_key(UserName=customer_user) |
| 129 | access_key_id = access_key["AccessKey"]["AccessKeyId"] |
| 130 | secret_access_key = access_key["AccessKey"]["SecretAccessKey"] |
| 131 | |
| 132 | try: |
| 133 | # Creating a connection with those credentials should work. |
| 134 | c.sql( |
| 135 | f""" |
| 136 | CREATE SECRET aws_secret_access_key AS '{secret_access_key}'; |
| 137 | CREATE CONNECTION aws_credentials TO AWS ( |
| 138 | ACCESS KEY ID = '{access_key_id}', |
| 139 | SECRET ACCESS KEY = SECRET aws_secret_access_key |
| 140 | ); |
| 141 | """, |
| 142 | print_statement=False, |
| 143 | ) |
| 144 | # Wait for IAM to propagate. |
| 145 | c.sleep(ctx.iam_propagation_seconds) |
| 146 | c.sql("VALIDATE CONNECTION aws_credentials") |
| 147 | |
| 148 | # Corrupting the secret access key should cause authentication to fail with |
| 149 | # an invalid signature error. |
| 150 | bad_secret_access_key = codecs.encode(secret_access_key, "rot13") |
| 151 | c.sql( |
| 152 | f"ALTER SECRET aws_secret_access_key AS '{bad_secret_access_key}'", |
| 153 | print_statement=False, |
| 154 | ) |
| 155 | try: |
| 156 | c.sql("VALIDATE CONNECTION aws_credentials") |
| 157 | except SystemError as e: |
| 158 | assert ( |
| 159 | e.diag.message_primary |
| 160 | and "SignatureDoesNotMatch" in e.diag.message_primary |
| 161 | ), e |
| 162 | else: |
| 163 | raise RuntimeError("connection validation unexpectedly succeeded") |
| 164 | |
| 165 | # Changing the access key to a nonexistent access key should fail with an |
| 166 | # invalid client ID error. |
| 167 | c.sql( |
| 168 | "ALTER CONNECTION aws_credentials SET (ACCESS KEY ID = 'AKIAV2KIV5LP3RAKAZUY')", |
| 169 | print_statement=False, |
| 170 | ) |
| 171 | try: |
| 172 | c.sql("VALIDATE CONNECTION aws_credentials") |
| 173 | except SystemError as e: |
| 174 | assert ( |
| 175 | e.diag.message_primary |
| 176 | and "InvalidClientTokenId" in e.diag.message_primary |
| 177 | ), e |
| 178 | else: |
| 179 | raise RuntimeError("connection validation unexpectedly succeeded") |
| 180 | finally: |
| 181 | ctx.iam.delete_access_key(UserName=customer_user, AccessKeyId=access_key_id) |
nothing calls this directly
no test coverage detected