{{ message }}
Repository navigation
Expand file tree
/
Copy pathclone_access_modules.py
More file actions
262 lines (210 loc) · 8.6 KB
/
Copy pathclone_access_modules.py
File metadata and controls
262 lines (210 loc) · 8.6 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
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
""" Script to clone access modules from git urls specified in config.json """
import logging
import os
import re
import shutil
import sys
import time
from git import Repo, GitCommandError
from . import helpers
logger = logging.getLogger(__name__)
logging.basicConfig(
level=logging.INFO,
handlers=[
logging.StreamHandler()
]
)
def ensure_access_modules_config(config):
""" Validate access_modules config """
if "access_modules" not in config:
raise Exception(
"Not configured properly."
"Config is expected to have access_modules key"
)
if "git_urls" not in config["access_modules"]:
raise Exception(
"Not configured properly."
" Config is expected to have git_urls key"
" as an array of git repos for access modules"
)
def remove_stale_cloned_modules():
"""
Remove already present modules so that we can clone new ones
This is to ensure that we are on the correct commit
"""
for each_access_module in os.listdir('Access/access_modules'):
helpers.remove_directory_with_contents(
f"Access/access_modules/{each_access_module}")
def initialize_init_file():
"""
Initialize __init__.py file in access modules root directory
from a template. This is to ensure that the modules are loaded
properly
"""
shutil.copyfile(
'Access/base_email_access/access_modules_init.py',
"Access/access_modules/__init__.py"
)
def get_repo_url_and_branch(formatted_git_arg):
""" Get url and branch from formatted git url string """
url = formatted_git_arg
target_branch = None
if "#" in url:
target_branch = url.split("#")[1]
url = url.split("#")[0]
return url, target_branch
ACCESS_MODULES_DIR = "./Access/access_modules"
def safe_access_module_path(name):
"""Resolve ``./Access/access_modules/<name>`` while rejecting path traversal.
``name`` is untrusted: it is derived from a git URL (folder name) or read
from a cloned repository's directory listing. Strip any path components with
``os.path.basename`` and confirm the resolved path stays inside
``access_modules`` before it is used as a clone target or rename destination.
"""
safe_name = os.path.basename(str(name).strip().rstrip("/"))
if safe_name in ("", ".", ".."):
raise Exception(f"Unsafe access module name derived from {name!r}")
target_path = os.path.join(ACCESS_MODULES_DIR, safe_name)
base_dir = os.path.realpath(ACCESS_MODULES_DIR)
resolved = os.path.realpath(target_path)
if resolved != base_dir and not resolved.startswith(base_dir + os.sep):
raise Exception(f"Refusing path outside access_modules: {name!r}")
return target_path
def _pinned_commit_sha(target_ref):
"""Return target_ref when it is a full 40-char commit SHA (an integrity pin)."""
if target_ref and re.fullmatch(r"[0-9a-fA-F]{40}", target_ref):
return target_ref
return None
def verify_module_integrity(repo, url, pinned_sha):
"""F-019: verify a cloned module against its pinned commit SHA.
External modules are cloned at runtime and imported directly, so a
compromised upstream = RCE. If a commit SHA was pinned, check it out and
assert HEAD matches, aborting on mismatch; otherwise warn that integrity
could not be verified.
"""
if not pinned_sha:
logger.warning(
"Access module %s is not pinned to a commit SHA; runtime integrity "
"cannot be verified (F-019). Pin via '<git-url>#<40-char-commit-sha>'.",
url,
)
return
repo.git.checkout(pinned_sha)
actual_sha = repo.head.commit.hexsha
if actual_sha.lower() != pinned_sha.lower():
raise Exception(
f"Integrity check failed for {url}: expected commit "
f"{pinned_sha}, got {actual_sha}"
)
logger.info("Verified access module %s at pinned commit %s", url, pinned_sha)
def clone_repo(formatted_git_arg, retry_limit):
""" Clone a single repo """
url, target_ref = get_repo_url_and_branch(formatted_git_arg)
folder_name = url.split("/").pop()[:-4]
# F-023: folder_name comes from the (untrusted) git URL — sanitize before use
target_folder_path = safe_access_module_path(folder_name)
# F-019: a full 40-char SHA after '#' is an integrity pin, not a branch.
pinned_sha = _pinned_commit_sha(target_ref)
retry_exception = None
repo = None
for clone_attempt in range(1, retry_limit + 1):
try:
logger.info("Cloning Repo")
if target_ref and not pinned_sha:
repo = Repo.clone_from(url, target_folder_path, branch=target_ref)
else:
repo = Repo.clone_from(url, target_folder_path)
except (GitCommandError, Exception) as exception:
sleep_time = 10 * clone_attempt
logger.error(
"Error while cloning repo. Error %s.",
exception,
)
logger.info(
"Retrying cloning: %s/%s. Backoff sleep %s",
clone_attempt,
retry_limit,
sleep_time,
)
retry_exception = exception
time.sleep(sleep_time)
else:
retry_exception = None
break
if retry_exception is not None:
logger.exception("Max retry count reached while cloning repo")
raise retry_exception
verify_module_integrity(repo, url, pinned_sha)
return target_folder_path
def move_modules_from_cloned_repo(cloned_path):
"""
Move access modules from the cloned folder to proper structure
"""
for each_path in os.listdir(cloned_path):
blacklist_paths = [
".git",
".github",
"secrets",
"docs",
]
if (
os.path.isdir(cloned_path + "/" + each_path)
and each_path not in blacklist_paths
):
try:
# F-021: each_path comes from the cloned repo — sanitize the
# rename destination so a crafted dir name can't escape.
os.rename(
cloned_path + "/" + each_path,
safe_access_module_path(each_path)
)
except Exception as exception:
logger.exception(
"Failed to move access module to proper structure %s",
exception
)
raise Exception(
f"Failed to move access module to proper structure {exception}") from exception
def ensure_access_modules_requirements(
cloned_path, core_requirements_file_path):
""" Consolidate requirements from multiple access modules """
for each_path in os.listdir(cloned_path):
if each_path == "requirements.txt":
current_requirements_file = cloned_path + "/" + each_path
all_requirements = helpers.read_content_from_file(
current_requirements_file)
all_requirements.extend(
helpers.read_content_from_file(core_requirements_file_path))
# Ensure requirements are unique
merged_requirements = list(set(all_requirements))
# Update the requirements.txt
helpers.write_content_to_file(
core_requirements_file_path, sorted(merged_requirements))
def clone_access_modules():
""" Core function to clone access modules repo """
config = helpers.read_json_from_file("./config.json")
ensure_access_modules_config(config)
retry_limit = config["access_modules"].get("RETRY_LIMIT", 5)
git_urls = config["access_modules"]["git_urls"]
requirements_file_path = 'Access/access_modules/requirements.txt'
helpers.ensure_folder_exists('Access/access_modules')
remove_stale_cloned_modules()
initialize_init_file()
helpers.ensure_file_exists(requirements_file_path)
# Ensure cleanup of requirements file before starting cloning process
helpers.write_content_to_file(requirements_file_path, [])
for formatted_git_arg in git_urls:
cloned_path = clone_repo(formatted_git_arg, retry_limit)
ensure_access_modules_requirements(cloned_path, requirements_file_path)
move_modules_from_cloned_repo(cloned_path)
logger.info("Cloning successful!")
helpers.remove_directory_with_contents(cloned_path)
def __main__():
logger.info("Starting cloning setup")
try:
clone_access_modules()
except Exception as exception:
logger.exception("Access module cloning failed!")
logger.exception(exception)
sys.exit(1)
__main__()
You can’t perform that action at this time.