Coverage for chalicelib/core/sourcemaps/sourcemaps.py: 10%
148 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 12:56 +0000
1import logging
2from urllib.parse import urlparse
4import requests
5from decouple import config
7from chalicelib.core.sourcemaps import sourcemaps_parser
8from chalicelib.utils import ssrf
9from chalicelib.utils.log import sanitize
10from chalicelib.utils.storage import StorageClient, generators
12logger = logging.getLogger(__name__)
14# When disabled, sourcemaps are only read from the bucket, never fetched
15# from the origin server referenced by error stack frames.
16FETCH_SOURCEMAPS_FROM_URL = config("FETCH_SOURCEMAPS_FROM_URL", cast=bool, default=True)
19def presign_share_urls(project_id, urls):
20 results = []
21 for u in urls:
22 results.append(StorageClient.get_presigned_url_for_sharing(bucket=config('sourcemaps_bucket'), expires_in=120,
23 key=generators.generate_file_key_from_url(project_id, u),
24 check_exists=True))
25 return results
28def presign_upload_urls(project_id, urls):
29 results = []
30 for u in urls:
31 results.append(StorageClient.get_presigned_url_for_upload(bucket=config('sourcemaps_bucket'),
32 expires_in=1800,
33 key=generators.generate_file_key_from_url(project_id, u)))
34 return results
37def __format_frame_old(f):
38 if f.get("context") is None:
39 f["context"] = []
40 else:
41 f["context"] = [[f["line"], f["context"]]]
42 url = f.pop("url")
43 f["absPath"] = url
44 f["filename"] = urlparse(url).path
45 f["lineNo"] = f.pop("line")
46 f["colNo"] = f.pop("column")
47 f["function"] = f.pop("func")
48 return f
51def __frame_is_valid(f):
52 return "columnNumber" in f and \
53 "lineNumber" in f and \
54 "fileName" in f
57def __format_frame(f):
58 f["context"] = [] # no context by default
59 if "source" in f:
60 f.pop("source")
61 url = f.pop("fileName")
62 f["absPath"] = url
63 f["filename"] = urlparse(url).path
64 f["lineNo"] = f.pop("lineNumber")
65 f["colNo"] = f.pop("columnNumber")
66 f["function"] = f.pop("functionName") if "functionName" in f else None
67 return f
70def format_payload(p, truncate_to_first=False):
71 if type(p) is list:
72 return [__format_frame(f) for f in (p[:1] if truncate_to_first else p) if __frame_is_valid(f)]
73 if type(p) is dict:
74 stack = p.get("stack", [])
75 return [__format_frame_old(f) for f in (stack[:1] if truncate_to_first else stack)]
76 return []
79def url_exists(url):
80 if not FETCH_SOURCEMAPS_FROM_URL:
81 return False
82 # frame URLs originate from untrusted tracker payloads; without this guard
83 # this HEAD (and the follow-up GET in sourcemapreader) is an SSRF primitive
84 if not ssrf.is_safe_external_url(url):
85 logger.warning(f"blocked unsafe sourcemap URL: {sanitize(url)}")
86 return False
87 try:
88 r = requests.head(url, allow_redirects=False,
89 timeout=config("sourcemapTimeout", cast=int, default=5))
90 return r.status_code == 200 and "text/html" not in r.headers.get("Content-Type", "")
91 except Exception as e:
92 logger.warning(f"!! Issue checking if URL exists: {sanitize(url)}")
93 logger.warning(sanitize(str(e)))
94 return False
97def get_traces_group(project_id, payload):
98 frames = format_payload(payload)
100 results = [{}] * len(frames)
101 payloads = {}
102 all_exists = True
103 for i, u in enumerate(frames):
104 file_exists_in_bucket = False
105 file_exists_in_server = False
106 file_url = u["absPath"]
107 key = generators.generate_file_key_from_url(project_id, file_url) # use filename instead?
108 params_idx = file_url.find("?")
109 if file_url and len(file_url) > 0 \
110 and not (file_url[:params_idx] if params_idx > -1 else file_url).endswith(".js"):
111 logger.debug(f"{sanitize(u['absPath'])} sourcemap is not a JS file")
112 payloads[key] = None
114 if key not in payloads:
115 file_exists_in_bucket = len(file_url) > 0 and StorageClient.exists(config('sourcemaps_bucket'), key)
116 if len(file_url) > 0 and not file_exists_in_bucket:
117 logger.debug(f"{sanitize(u['absPath'])} sourcemap (key '{sanitize(key)}') doesn't exist in S3 looking in server")
118 if not file_url.endswith(".map"):
119 file_url += '.map'
120 file_exists_in_server = url_exists(file_url)
121 file_exists_in_bucket = file_exists_in_server
122 all_exists = all_exists and file_exists_in_bucket
123 if not file_exists_in_bucket and not file_exists_in_server:
124 logger.debug(f"{sanitize(u['absPath'])} sourcemap (key '{sanitize(key)}') doesn't exist in S3 nor server")
125 payloads[key] = None
126 else:
127 payloads[key] = []
128 results[i] = dict(u)
129 results[i]["frame"] = dict(u)
130 if payloads[key] is not None:
131 payloads[key].append({"resultIndex": i, "frame": dict(u), "URL": file_url,
132 "position": {"line": u["lineNo"], "column": u["colNo"]},
133 "isURL": file_exists_in_server})
135 for key in payloads.keys():
136 if payloads[key] is None:
137 continue
138 key_results = sourcemaps_parser.get_original_trace(
139 key=payloads[key][0]["URL"] if payloads[key][0]["isURL"] else key,
140 positions=[o["position"] for o in payloads[key]],
141 is_url=payloads[key][0]["isURL"])
142 if key_results is None:
143 all_exists = False
144 continue
145 for i, r in enumerate(key_results):
146 res_index = payloads[key][i]["resultIndex"]
147 # function name search by frontend lib is better than sourcemaps' one in most cases
148 if results[res_index].get("function") is not None:
149 r["function"] = results[res_index]["function"]
150 r["frame"] = payloads[key][i]["frame"]
151 results[res_index] = r
152 return fetch_missed_contexts(results), all_exists
155def get_js_cache_path(fullURL):
156 p = urlparse(fullURL)
157 return p.scheme + '/' + p.netloc + p.path # TODO (Also in go assets library): What if URL with query? (like versions)
160MAX_COLUMN_OFFSET = 60
163def fetch_missed_contexts(frames):
164 source_cache = {}
165 for i in range(len(frames)):
166 if frames[i] and frames[i].get("context") and len(frames[i]["context"]) > 0:
167 continue
168 file_abs_path = frames[i]["frame"]["absPath"]
169 if file_abs_path in source_cache:
170 file = source_cache[file_abs_path]
171 else:
172 file_path = get_js_cache_path(file_abs_path)
173 file = StorageClient.get_file(config('js_cache_bucket'), file_path)
174 if file is None:
175 logger.debug(f"Missing abs_path: {sanitize(file_abs_path)}, file {sanitize(file_path)} not found in {config('js_cache_bucket')}")
176 source_cache[file_abs_path] = file
177 if file is None:
178 continue
179 lines = file.split("\n")
181 if frames[i]["lineNo"] is None:
182 logger.debug("no original-source found for frame in sourcemap results")
183 frames[i] = frames[i]["frame"]
184 frames[i]["originalMapping"] = False
186 l = frames[i]["lineNo"] - 1 # starts from 1
187 c = frames[i]["colNo"] - 1 # starts from 1
188 if len(lines) == 1:
189 logger.debug("minified asset")
190 l = frames[i]["frame"]["lineNo"] - 1 # starts from 1
191 c = frames[i]["frame"]["colNo"] - 1 # starts from 1
192 elif l >= len(lines):
193 logger.debug(f"line number {l} greater than file length {len(lines)}")
194 continue
196 line = lines[l]
197 offset = c - MAX_COLUMN_OFFSET
198 if offset < 0: # if the line is short
199 offset = 0
200 frames[i]["context"].append([frames[i]["lineNo"], line[offset: c + MAX_COLUMN_OFFSET + 1]])
201 return frames