python: Inline expectation should have space after $

This was a regex-find-replace from `# \$(?! )` (using a negative lookahead) to `# $ `.

Author: owen-mc

python/ql/test/experimental/query-tests/Security/CWE-022-UnsafeUnpacking/UnsafeUnpack.py

@@ -1,6 +1,6 @@
 import requests
 import shutil
-import os 
+import os
 
 from flask import Flask, request
 app = Flask(__name__)
@@ -16,8 +16,8 @@ def download_from_url():
             with open(tarpath, "wb") as f:
                   f.write(response.raw.read())
             untarredpath = "/tmp/tmp123"
-            shutil.unpack_archive(tarpath, untarredpath)  # $result=BAD
-        
+            shutil.unpack_archive(tarpath, untarredpath)  # $ result=BAD
+
 
 # A source catching an S3 filename download
 # see boto3: https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/s3.html#S3.Client.download_file
@@ -31,7 +31,7 @@ def download_from_url():
 
 s3 = boto3.client('s3')
 s3.download_file(bucket_name, remote_ziped_name, local_ziped_path)
-shutil.unpack_archive(local_ziped_path, base_dir)  # $result=BAD
+shutil.unpack_archive(local_ziped_path, base_dir)  # $ result=BAD
 
 
 # wget
@@ -45,11 +45,11 @@ def download_from_url():
 
 # download(url, out, bar) contains out parameter
 wget.download(url, compressed_file)
-shutil.unpack_archive(compressed_file, base_dir) # $result=BAD
+shutil.unpack_archive(compressed_file, base_dir) # $ result=BAD
 
 # download(url) returns filename
 compressed_file = wget.download(url)
-shutil.unpack_archive(compressed_file, base_dir) # $result=BAD
+shutil.unpack_archive(compressed_file, base_dir) # $ result=BAD
 
 
 # A source coming from a CLI argparse module
@@ -63,7 +63,7 @@ def download_from_url():
 
 args = parser.parse_args()
 compressed_file = args.filename
-shutil.unpack_archive(compressed_file, base_dir) # $result=BAD
+shutil.unpack_archive(compressed_file, base_dir) # $ result=BAD
 
 
 # A source coming from a CLI and downloaded
@@ -83,8 +83,8 @@ def download_from_url():
 tarpath = "/tmp/tmp456/tarball.tar.gz"
 with open(tarpath, "wb") as f:
       f.write(response.raw.read())
-      
-shutil.unpack_archive(tarpath, base_dir) # $result=BAD
+
+shutil.unpack_archive(tarpath, base_dir) # $ result=BAD
 
 # the django upload functionality
 # see HttpRequest.FILES: https://docs.djangoproject.com/en/4.1/ref/request-response/#django.http.HttpRequest.FILES
@@ -97,36 +97,36 @@ def simple_upload(request):
       base_dir = "/tmp/baase_dir"
       if request.method == 'POST':
             # Read uploaded files by chunks of data
-            # see chunks(): https://docs.djangoproject.com/en/4.1/ref/files/uploads/#django.core.files.uploadedfile.UploadedFile.chunks 
+            # see chunks(): https://docs.djangoproject.com/en/4.1/ref/files/uploads/#django.core.files.uploadedfile.UploadedFile.chunks
             savepath = os.path.join(base_dir, "tarball_compressed.tar.gz")
             with open(savepath, 'wb+') as wfile:
                   for chunk in request.FILES["ufile1"].chunks():
                         wfile.write(chunk)
-            shutil.unpack_archive(savepath, base_dir) # $result=BAD
+            shutil.unpack_archive(savepath, base_dir) # $ result=BAD
 
             # Write in binary the uploaded tarball
             myfile = request.FILES.get("ufile1")
             file_path = os.path.join(base_dir, "tarball.tar")
             with file_path.open('wb') as f:
                   f.write(myfile.read())
-            shutil.unpack_archive(file_path, base_dir) # $result=BAD
+            shutil.unpack_archive(file_path, base_dir) # $ result=BAD
 
             # Save uploaded files using FileSystemStorage Django API
             # see FileSystemStorage: https://docs.djangoproject.com/en/4.1/ref/files/storage/#django.core.files.storage.FileSystemStorage
             for ufile in  request.FILES.getlist():
                   fs = FileSystemStorage()
                   filename = fs.save(ufile.name, ufile)
                   uploaded_file_path = fs.path(filename)
-                  shutil.unpack_archive(uploaded_file_path, base_dir) # $result=BAD
-            
+                  shutil.unpack_archive(uploaded_file_path, base_dir) # $ result=BAD
+
             return render(request, 'simple_upload.html')
 
       elif request.method == 'GET':
             return render(request, 'simple_upload.html')
 
 
 import shutil
-import os 
+import os
 import tarfile
 import tempfile
 import argparse
@@ -139,8 +139,8 @@ def simple_upload(request):
 args = parser.parse_args()
 unsafe_filename_tar = args.filename
 with tarfile.TarFile(unsafe_filename_tar, mode="r") as tar:
-    tar.extractall(path="/tmp/unpack/", members=tar) # $result=BAD
-tar = tarfile.open(unsafe_filename_tar)   
+    tar.extractall(path="/tmp/unpack/", members=tar) # $ result=BAD
+tar = tarfile.open(unsafe_filename_tar)
 
 
 from django.shortcuts import render
@@ -152,19 +152,19 @@ def simple_upload(request):
       base_dir = "/tmp/baase_dir"
       if request.method == 'POST':
             # Read uploaded files by chunks of data
-            # see chunks(): https://docs.djangoproject.com/en/4.1/ref/files/uploads/#django.core.files.uploadedfile.UploadedFile.chunks 
+            # see chunks(): https://docs.djangoproject.com/en/4.1/ref/files/uploads/#django.core.files.uploadedfile.UploadedFile.chunks
             savepath = os.path.join(base_dir, "tarball_compressed.tar.gz")
             with open(savepath, 'wb+') as wfile:
                   for chunk in request.FILES["ufile1"].chunks():
                         wfile.write(chunk)
 
                   tar = tarfile.open(savepath)
                   result = []
-                  for member in tar:      
-                      if member.issym():  
-                          raise ValueError("But it is a symlink")       
-                      result.append(member)     
-                  tar.extractall(path=tempfile.mkdtemp(), members=result) # $result=BAD   
+                  for member in tar:
+                      if member.issym():
+                          raise ValueError("But it is a symlink")
+                      result.append(member)
+                  tar.extractall(path=tempfile.mkdtemp(), members=result) # $ result=BAD
                   tar.close()
 
 
@@ -173,7 +173,7 @@ def simple_upload(request):
 with open(tarpath, "wb") as f:
       f.write(response.raw.read())
 target_dir = "/tmp/unpack"
-tarfile.TarFile(tarpath, mode="r").extractall(path=target_dir) # $result=BAD
+tarfile.TarFile(tarpath, mode="r").extractall(path=target_dir) # $ result=BAD
 
 
 from pathlib import Path
@@ -183,7 +183,7 @@ def simple_upload(request):
 def default_session() -> boto3.Session:
     _SESSION = None
     if _SESSION is None:
-        _SESSION = boto3.Session() 
+        _SESSION = boto3.Session()
     return _SESSION
 
 cache = False
@@ -198,4 +198,4 @@ def default_session() -> boto3.Session:
           target = cache_dir
       else:
           target = Path(tempfile.mkdtemp())
-      shutil.unpack_archive(tmp.name, target) # $result=BAD
+      shutil.unpack_archive(tmp.name, target) # $ result=BAD

python/ql/test/experimental/query-tests/Security/CWE-1427-PromptInjection/agent_instructions.py

@@ -6,7 +6,7 @@
 def get_input1():
     input = request.args.get("input")
 
-    agent = Agent(name="Assistant", instructions="This prompt is customized for " + input) # $Alert[py/prompt-injection]
+    agent = Agent(name="Assistant", instructions="This prompt is customized for " + input) # $ Alert[py/prompt-injection]
 
     result = Runner.run_sync(agent, "This is a user message.")
     print(result.final_output)
@@ -22,17 +22,17 @@ def get_input2():
         input=[
             {
                 "role": "user",
-                "content": input, # $Alert[py/prompt-injection]
+                "content": input, # $ Alert[py/prompt-injection]
             }
-        ] 
+        ]
     )
 
     result2 = Runner.run_sync(
         agent,
         [
             {
                 "role": "user",
-                "content": input, # $Alert[py/prompt-injection]
+                "content": input, # $ Alert[py/prompt-injection]
             }
-        ] 
+        ]
     )

python/ql/test/experimental/query-tests/Security/CWE-176/samples.py

@@ -7,7 +7,7 @@
 @app.route("/unsafe1")
 def unsafe1():
     user_input = escape(request.args.get("ui"))
-    normalized_user_input = unicodedata.normalize("NFKC", user_input)  # $result=BAD
+    normalized_user_input = unicodedata.normalize("NFKC", user_input)  # $ result=BAD
     return render_template("result.html", normalized_user_input=normalized_user_input)
 
 
@@ -17,14 +17,14 @@ def unsafe1bis():
     if user_input.isascii():
         normalized_user_input = user_input
     else:
-        normalized_user_input = unicodedata.normalize("NFC", user_input)  # $result=BAD
+        normalized_user_input = unicodedata.normalize("NFC", user_input)  # $ result=BAD
     return render_template("result.html", normalized_user_input=normalized_user_input)
 
 
 @app.route("/safe1")
 def safe1():
     normalized_user_input = unicodedata.normalize(
         "NFKC", request.args.get("ui")
-    )  # $result=OK
+    )  # $ result=OK
     user_input = escape(normalized_user_input)
     return render_template("result.html", normalized_user_input=user_input)

python/ql/test/library-tests/dataflow/global-flow/test.py

@@ -2,15 +2,15 @@
 
 # Simple assignment
 
-g = [5] # $writes=g
+g = [5] # $ writes=g
 
 # Multiple assignment
 
-g1, g2 = [6], [7] # $writes=g1 writes=g2
+g1, g2 = [6], [7] # $ writes=g1 writes=g2
 
 # Assignment that's only referenced in this scope.
 
-unreferenced_g = [8] # $writes=unreferenced_g
+unreferenced_g = [8] # $ writes=unreferenced_g
 print(unreferenced_g)
 
 # Testing modifications of globals
@@ -24,98 +24,98 @@
 # but currently our analysis thinks `g_mod` might be used in the `print` call
 g_mod = [10] # $ SPURIOUS: writes=g_mod
 print("foo")
-g_mod = [100] # $writes=g_mod
+g_mod = [100] # $ writes=g_mod
 
 # Modification by mutation
 
-g_ins = [50] # $writes=g_ins
+g_ins = [50] # $ writes=g_ins
 print(g_ins)
 g_ins.append(75)
 
 # A global with multiple potential definitions
 
-import unknown_module # $writes=unknown_module
+import unknown_module # $ writes=unknown_module
 if unknown_module.attr:
-    g_mult = [200] # $writes=g_mult
+    g_mult = [200] # $ writes=g_mult
 else:
-    g_mult = [300] # $writes=g_mult
+    g_mult = [300] # $ writes=g_mult
 
 # A global variable that may be redefined depending on some unknown value
 
-g_redef = [400] # $writes=g_redef
+g_redef = [400] # $ writes=g_redef
 if unknown_module.attr:
-    g_redef = [500] # $writes=g_redef
+    g_redef = [500] # $ writes=g_redef
 
-def global_access(): # $writes=global_access
+def global_access(): # $ writes=global_access
     l = 5
-    print(g) # $reads=g
-    print(g1) # $reads=g1
-    print(g2) # $reads=g2
-    print(g_mod) # $reads=g_mod
-    print(g_ins) # $reads=g_ins
-    print(g_mult) # $reads=g_mult
-    print(g_redef) # $reads=g_redef
-
-def print_g_mod(): # $writes=print_g_mod
-    print(g_mod) # $reads=g_mod
-
-def global_mod(): # $writes=global_mod
+    print(g) # $ reads=g
+    print(g1) # $ reads=g1
+    print(g2) # $ reads=g2
+    print(g_mod) # $ reads=g_mod
+    print(g_ins) # $ reads=g_ins
+    print(g_mult) # $ reads=g_mult
+    print(g_redef) # $ reads=g_redef
+
+def print_g_mod(): # $ writes=print_g_mod
+    print(g_mod) # $ reads=g_mod
+
+def global_mod(): # $ writes=global_mod
     global g_mod
-    g_mod += [150] # $reads,writes=g_mod
-    print_g_mod() # $reads=print_g_mod
+    g_mod += [150] # $ reads,writes=g_mod
+    print_g_mod() # $ reads=print_g_mod
 
-def global_inside_local_function(): # $writes=global_inside_local_function
+def global_inside_local_function(): # $ writes=global_inside_local_function
     def local_function():
-        print(g) # $reads=g
+        print(g) # $ reads=g
     local_function()
 
 ## Imports
 
 
 # Direct imports
 
-import foo_module # $writes=foo_module
+import foo_module # $ writes=foo_module
 
-def use_foo(): # $writes=use_foo
-    print(foo_module.attr) # $reads=foo_module
+def use_foo(): # $ writes=use_foo
+    print(foo_module.attr) # $ reads=foo_module
 
 # Partial imports
 
-from bar import baz_attr, quux_attr # $writes=baz_attr writes=quux_attr
+from bar import baz_attr, quux_attr # $ writes=baz_attr writes=quux_attr
 
-def use_partial_import(): # $writes=use_partial_import
-    print(baz_attr, quux_attr) # $reads=baz_attr reads=quux_attr
+def use_partial_import(): # $ writes=use_partial_import
+    print(baz_attr, quux_attr) # $ reads=baz_attr reads=quux_attr
 
 # Aliased imports
 
-from spam_module import ham_attr as eggs_attr # $writes=eggs_attr
+from spam_module import ham_attr as eggs_attr # $ writes=eggs_attr
 
-def use_aliased_import(): # $writes=use_aliased_import
-    print(eggs_attr) # $reads=eggs_attr
+def use_aliased_import(): # $ writes=use_aliased_import
+    print(eggs_attr) # $ reads=eggs_attr
 
 # Import star (unlikely to work unless we happen to extract/model the referenced module)
 
 # Unknown modules
 
 from unknown import *
 
-def secretly_use_unknown(): # $writes=secretly_use_unknown
-    print(unknown_attr) # $reads=unknown_attr
+def secretly_use_unknown(): # $ writes=secretly_use_unknown
+    print(unknown_attr) # $ reads=unknown_attr
 
 # Known modules
 
 from known import *
 
-def secretly_use_known(): # $writes=secretly_use_known
-    print(known_attr) # $reads=known_attr
+def secretly_use_known(): # $ writes=secretly_use_known
+    print(known_attr) # $ reads=known_attr
 
 # Local import in function
 
-def imports_locally(): # $writes=imports_locally
+def imports_locally(): # $ writes=imports_locally
     import mod1
 
 # Global import hidden in function
 
-def imports_stuff(): # $writes=imports_stuff
+def imports_stuff(): # $ writes=imports_stuff
     global mod2
-    import mod2 # $writes=mod2
+    import mod2 # $ writes=mod2