-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathutilities.py
98 lines (86 loc) · 4.01 KB
/
utilities.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
# define function to retrieve PiX creds
import boto3
from botocore.exceptions import ClientError
import json
import base64
import os
import seleniumwire.webdriver as webdriver
from selenium.webdriver.chrome.options import Options as ChromeOptions
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.chrome.service import Service
import logging
def get_driver():
os.environ['WDM_LOG'] = '0'
logging.getLogger('seleniumwire').setLevel(logging.CRITICAL)
# Options for docker image
chrome_options = ChromeOptions()
chrome_options.add_argument("--headless")
chrome_options.add_argument("--no-sandbox")
chrome_options.add_argument("--disable-dev-shm-usage")
chrome_options.add_argument('--disable-blink-features=AutomationControlled')
chrome_options.add_experimental_option("prefs", {
"download.default_directory": os.getcwd(),
"download.directory_upgrade": True
})
chrome_driver_path = "/usr/bin/chromedriver"
service = Service(executable_path=chrome_driver_path)
print("Initializing...")
driver = webdriver.Chrome(service=service, options=chrome_options)
wait = WebDriverWait(driver, 30)
return driver, wait
def get_credentials(secret_name):
"""Retrieves and returns the username and password from local credentials file OR Secrets Manager.
Returns:
tuple: A tuple containing the username as the first element and
the password as the second element.
"""
try: # try to get locally
import credentials
username = credentials.forms_connection['username']
password = credentials.forms_connection['password']
return username, password
except:
region_name = "us-east-2"
# Create a Secrets Manager client
session = boto3.session.Session()
client = session.client(
service_name='secretsmanager',
region_name=region_name
)
try:
get_secret_value_response = client.get_secret_value(
SecretId=secret_name
)
except ClientError as e:
if e.response['Error']['Code'] == 'DecryptionFailureException':
# Secrets Manager can't decrypt the protected secret text using the provided KMS key.
# Deal with the exception here, and/or rethrow at your discretion.
raise e
elif e.response['Error']['Code'] == 'InternalServiceErrorException':
# An error occurred on the server side.
# Deal with the exception here, and/or rethrow at your discretion.
raise e
elif e.response['Error']['Code'] == 'InvalidParameterException':
# You provided an invalid value for a parameter.
# Deal with the exception here, and/or rethrow at your discretion.
raise e
elif e.response['Error']['Code'] == 'InvalidRequestException':
# You provided a parameter value that is not valid for the current state of the resource.
# Deal with the exception here, and/or rethrow at your discretion.
raise e
elif e.response['Error']['Code'] == 'ResourceNotFoundException':
# We can't find the resource that you asked for.
# Deal with the exception here, and/or rethrow at your discretion.
raise e
else:
# Decrypts secret using the associated KMS CMK.
# Depending on whether the secret is a string or binary, one of these fields will be populated.
if 'SecretString' in get_secret_value_response:
secret = get_secret_value_response['SecretString']
secret = json.loads(secret) # Assuming JSON format
else:
decoded_binary_secret = base64.b64decode(get_secret_value_response['SecretBinary'])
secret = json.loads(decoded_binary_secret) # Assuming JSON format
return secret