Skip to content

Bump bandit from 1.7.4 to 1.8.6 #3205

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 1 commit into from
Oct 29, 2025
Merged

Bump bandit from 1.7.4 to 1.8.6 #3205

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion requirements.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -104,7 +104,8 @@ ctlsettings==0.4.3
pbr==7.0.0
PyYAML==6.0.2
stevedore==5.5.0
bandit==1.7.4
rich==14.2.0 # bandit
bandit==1.8.6

defusedxml==0.7.1
lxml==6.0.0
Expand Down
20 changes: 10 additions & 10 deletions wardenclyffe/main/tests/test_views.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -681,7 +681,7 @@ def test_video_add_file_form(self):
"Token" : "2336412f37fb687f5d51e6e241d09",
"TopicArn" : "arn:aws:sns:us-east-1:123456789012:MyTopic",
"Message" : "You have chosen to subscribe to the topic arn:aws:sns:",
"SubscribeURL" : "http://example.com/?Action=ConfirmSubscription",
"SubscribeURL" : "https://example.com/?Action=ConfirmSubscription",
"Timestamp" : "2012-04-26T20:45:04.751Z",
"SignatureVersion" : "1",
"Signature" : "EXAMPLEpH+DcEwjAPg8O9mY8dReBSwksfg2S7WKQcikcNK+gLPoBc1Q=",
Expand All @@ -705,7 +705,7 @@ def setUp(self):
def test_subscription_confirmation(self):
HTTPretty.register_uri(
HTTPretty.GET,
"http://example.com/?Action=ConfirmSubscription",
"https://example.com/?Action=ConfirmSubscription",
body="yes"
)
r = self.c.post(
Expand Down Expand Up @@ -856,11 +856,11 @@ def test_rows(self):
f = FileFactory()
FileFactory(location_type='panopto', video=f.video, filename='alpha')
FileFactory(location_type='youtube', video=f.video,
url='http://www.youtube.com/watch?v=fS4qBPdhr8A')
url='https://www.youtube.com/watch?v=fS4qBPdhr8A')

with self.settings(
PANOPTO_LINK_URL='http://testserver/link/{}/',
PANOPTO_EMBED_URL='http://testserver/embed/{}/'):
PANOPTO_LINK_URL='https://testserver/link/{}/',
PANOPTO_EMBED_URL='https://testserver/embed/{}/'):
view = CollectionReportView()
rows = view.rows(f.video.collection)
self.assertEqual(len(rows), 1)
Expand All @@ -876,12 +876,12 @@ def test_rows(self):
("56d27944-4131-11e1-8164-0017f20ea192"
"-Mediathread_video_uploaded_by_mlp55.mp4"))
self.assertEqual(rows[0][5], 'alpha')
self.assertEqual(rows[0][6], 'http://testserver/link/alpha/')
self.assertEqual(rows[0][7], 'http://testserver/embed/alpha/')
self.assertEqual(rows[0][6], 'https://testserver/link/alpha/')
self.assertEqual(rows[0][7], 'https://testserver/embed/alpha/')
self.assertEqual(
rows[0][8], 'http://www.youtube.com/watch?v=fS4qBPdhr8A')
rows[0][8], 'https://www.youtube.com/watch?v=fS4qBPdhr8A')
self.assertEqual(
rows[0][9], 'http://www.youtube.com/watch?v=fS4qBPdhr8A')
rows[0][9], 'https://www.youtube.com/watch?v=fS4qBPdhr8A')


class MockSftpStatAttrs(object):
Expand All @@ -904,7 +904,7 @@ def test_get_context_data_youtube(self):
cuit = FileFactory()
FileFactory(
location_type='youtube', video=cuit.video,
url='http://www.youtube.com/watch?v=fS4qBPdhr8A')
url='https://www.youtube.com/watch?v=fS4qBPdhr8A')

view = SureLinkVideoView()
view.request = RequestFactory().get('/', {'file': cuit.filename})
Expand Down
7 changes: 6 additions & 1 deletion wardenclyffe/main/views.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1378,7 +1378,12 @@
if "SubscribeURL" not in message:
return HttpResponse("no subscribe url", status=400)
url = message["SubscribeURL"]
r = requests.get(url)
parsed_url = urlparse(url)

if parsed_url.scheme != 'https':
return HttpResponse('invalid subscribe url', status=400)

r = requests.get(url, timeout=5)

Check failure

Code scanning / CodeQL

Full server-side request forgery Critical

The full URL of this request depends on a
user-provided value
Loading
.

Copilot Autofix

AI 6 days ago

To fix this issue robustly, the SubscribeURL extracted from SNS confirmation messages must not be blindly trusted; instead, the URL should be validated to ensure it only references domains legitimately controlled by AWS SNS (or whatever specific infrastructure is expected). This means checking the hostname against a whitelist of known prefix or suffixes (e.g., .amazonaws.com). A recommended approach is:

  • Parse the URL.
  • Check that its hostname ends with a trusted AWS SNS domain suffix (e.g., .amazonaws.com), optionally even more specific such as sns.<region>.amazonaws.com.
  • Consider using ipaddress and socket to resolve and ensure the endpoint is not private (though AWS endpoints tend to be public).
    Make these checks in the _subscription_confirmation method before calling requests.get(url, timeout=5).

Minimal changes should be made: add validation logic just before making the request, and ensure any required imports (import socket) are added if needed.

Suggested changeset 1
wardenclyffe/main/views.py

Autofix patch

Autofix patch
Run the following command in your local git repository to apply this patch
cat << 'EOF' | git apply
diff --git a/wardenclyffe/main/views.py b/wardenclyffe/main/views.py
--- a/wardenclyffe/main/views.py
+++ b/wardenclyffe/main/views.py
@@ -50,6 +50,7 @@
     from urllib.parse import urlparse
 except ImportError:
     from urlparse import urlparse
+import socket
 
 
 def is_staff(user):
@@ -1383,6 +1384,12 @@
         if parsed_url.scheme != 'https':
             return HttpResponse('invalid subscribe url', status=400)
 
+        # Check the hostname is an AWS SNS endpoint
+        allowed_domains = ('.amazonaws.com',)
+        hostname = parsed_url.hostname
+        if not hostname or not hostname.endswith(allowed_domains):
+            return HttpResponse('untrusted subscribe url', status=400)
+
         r = requests.get(url, timeout=5)
         if r.status_code == 200:
             return HttpResponse("OK")
EOF
@@ -50,6 +50,7 @@
from urllib.parse import urlparse
except ImportError:
from urlparse import urlparse
import socket


def is_staff(user):
@@ -1383,6 +1384,12 @@
if parsed_url.scheme != 'https':
return HttpResponse('invalid subscribe url', status=400)

# Check the hostname is an AWS SNS endpoint
allowed_domains = ('.amazonaws.com',)
hostname = parsed_url.hostname
if not hostname or not hostname.endswith(allowed_domains):
return HttpResponse('untrusted subscribe url', status=400)

r = requests.get(url, timeout=5)
if r.status_code == 200:
return HttpResponse("OK")
Copilot is powered by AI and may make mistakes. Always verify output.
if r.status_code == 200:
return HttpResponse("OK")
return HttpResponse("Failed to confirm")
Expand Down
4 changes: 2 additions & 2 deletions wardenclyffe/mediathread/tasks.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -96,7 +96,7 @@ def submit_to_mediathread(operation):
audio, width, height
)

r = requests.post(mediathread_base + "save/", params)
r = requests.post(mediathread_base + "save/", params, timeout=5)
if r.status_code == 200:
# requests follows redirects, so we need to get the location
# out of the history
Expand Down Expand Up @@ -136,7 +136,7 @@ def update_mediathread(operation):

params = mediathread_update_params(video, mediathread_secret)

r = requests.post(mediathread_base + 'update/', params)
r = requests.post(mediathread_base + 'update/', params, timeout=5)
if r.status_code == 200:
return ("complete", "")
elif r.status_code == 404:
Expand Down
2 changes: 1 addition & 1 deletion wardenclyffe/mediathread/views.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -109,7 +109,7 @@ class MediathreadCourseGetter(object):
def run(self, username):
try:
url = mediathread_url(username)
r = requests.get(url)
r = requests.get(url, timeout=5)
courses = r.json()['courses']
courses = [dict(id=k, title=v['title'])
for (k, v) in courses.items()]
Expand Down