forked from seung-lab/python-task-queue
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathsecrets.py
More file actions
118 lines (88 loc) · 3.5 KB
/
Copy pathsecrets.py
File metadata and controls
118 lines (88 loc) · 3.5 KB
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
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
from __future__ import print_function
from collections import defaultdict
import os
import json
from google.oauth2 import service_account
from .lib import mkdir, colorize
HOME = os.path.expanduser('~')
CLOUD_VOLUME_DIR = mkdir(os.path.join(HOME, '.cloudvolume', 'secrets'))
CLOUD_FILES_DIR = mkdir(os.path.join(HOME, '.cloudfiles', 'secrets'))
AWS_DEFAULT_REGION = 'us-east-1' if 'AWS_DEFAULT_REGION' not in os.environ else os.environ['AWS_DEFAULT_REGION']
def secretpath(filepath):
preferred = os.path.join(CLOUD_VOLUME_DIR, filepath)
if os.path.exists(preferred):
return preferred
backcompat = [
'/', # original
CLOUD_FILES_DIR,
]
backcompat = [ os.path.join(path, filepath) for path in backcompat ]
for path in backcompat:
if os.path.exists(path):
return path
return preferred
def default_google_project_name():
default_credentials_path = secretpath('google-secret.json')
if os.path.exists(default_credentials_path):
with open(default_credentials_path, 'rt') as f:
return json.loads(f.read())['project_id']
return None
PROJECT_NAME = default_google_project_name()
GOOGLE_CREDENTIALS_CACHE = {}
google_credentials_path = secretpath('google-secret.json')
def google_credentials(bucket = ''):
global PROJECT_NAME
global GOOGLE_CREDENTIALS_CACHE
if bucket in GOOGLE_CREDENTIALS_CACHE.keys():
return GOOGLE_CREDENTIALS_CACHE[bucket]
paths = [
secretpath('google-secret.json')
]
if bucket:
paths = [ secretpath('{}-google-secret.json'.format(bucket)) ] + paths
google_credentials = None
project_name = PROJECT_NAME
for google_credentials_path in paths:
if os.path.exists(google_credentials_path):
google_credentials = service_account.Credentials \
.from_service_account_file(google_credentials_path)
with open(google_credentials_path, 'rt') as f:
project_name = json.loads(f.read())['project_id']
break
if google_credentials == None:
print(colorize('yellow', 'Using default Google credentials. There is no ~/.cloudvolume/secrets/google-secret.json set.'))
else:
GOOGLE_CREDENTIALS_CACHE[bucket] = (project_name, google_credentials)
return project_name, google_credentials
AWS_CREDENTIALS_CACHE = defaultdict(dict)
aws_credentials_path = secretpath('aws-secret.json')
def aws_credentials(bucket = '', service = 'aws'):
global AWS_CREDENTIALS_CACHE
if service == 's3':
service = 'aws'
if bucket in AWS_CREDENTIALS_CACHE.keys():
return AWS_CREDENTIALS_CACHE[bucket]
default_file_path = '{}-secret.json'.format(service)
paths = [
secretpath(default_file_path)
]
if bucket:
paths = [ secretpath('{}-{}-secret.json'.format(bucket, service)) ] + paths
aws_credentials = {}
aws_credentials_path = secretpath(default_file_path)
for aws_credentials_path in paths:
if os.path.exists(aws_credentials_path):
with open(aws_credentials_path, 'r') as f:
aws_credentials = json.loads(f.read())
break
if not aws_credentials:
# did not find any secret json file, will try to find it in environment variables
if 'AWS_ACCESS_KEY_ID' in os.environ and 'AWS_SECRET_ACCESS_KEY' in os.environ:
aws_credentials = {
'AWS_ACCESS_KEY_ID': os.environ['AWS_ACCESS_KEY_ID'],
'AWS_SECRET_ACCESS_KEY': os.environ['AWS_SECRET_ACCESS_KEY'],
}
if 'AWS_DEFAULT_REGION' in os.environ:
aws_credentials['AWS_DEFAULT_REGION'] = os.environ['AWS_DEFAULT_REGION']
AWS_CREDENTIALS_CACHE[service][bucket] = aws_credentials
return aws_credentials