Added XXE module

astra.py

@@ -24,6 +24,7 @@
 from modules.sqli import sqli_check
 from modules.xss import xss_check
 from modules.redirect import open_redirect_check
+from modules.xxe import xxe_scan
 from core.zap_config import zap_start
 from multiprocessing import Process
 from utils.db import Database_update
@@ -32,7 +33,7 @@
 if os.getcwd().split('/')[-1] != 'API':
     from API.api import main
 
-
+xxe = xxe_scan()
 dbupdate = Database_update()
 
 def parse_collection(collection_name,collection_type):
@@ -145,7 +146,11 @@ def modules_scan(url,method,headers,body,scanid=None):
         update_scan_status(scanid, "xss")
     if attack['open-redirection'] == 'Y' or attack['open-redirection'] == 'y':
         open_redirect_check(url,method,headers,body,scanid)
-        update_scan_status(scanid, "open-redirection") 
+        update_scan_status(scanid, "open-redirection")
+    if attack['xxe'] == 'Y' or attack['xxe'] == 'y':
+        xxe.xxe_test(url,method,headers,body,scanid)
+        update_scan_status(scanid, "xxe") 
+
 
 def validate_data(url,method):
     ''' Validate HTTP request data and return boolean value'''
@@ -218,7 +223,6 @@ def scan_core(collection_type,collection_name,url,headers,method,body,loginurl,l
 
     else:
         print "%s [-]Invalid Collection. Please recheck collection Type/Name %s" %(api_logger.G, api_logger.W)
-    #generate_report()
 
 def get_arg(args=None):
         parser = argparse.ArgumentParser(description='Astra - REST API Security testing Framework')

modules/xxe.py

@@ -0,0 +1,109 @@
+import socket
+import sys
+import argparse
+import requests
+import threading
+import time
+import hashlib
+import os
+import sendrequest as req
+import utils.logger as logger
+import utils.logs as logs
+
+from utils.db import Database_update
+
+dbupdate = Database_update()
+
+# Dummy response
+data = b'''\
+HTTP/1.1 200 OK\r\n\
+Connection: close\r\n\
+Content-Type: text/html\r\n\
+Content-Length: 6\r\n\
+\r\n\
+Hello!\
+'''
+
+class xxe_scan:
+    def __init__(self):
+        self.port = 1111
+        self.host = socket.gethostbyname(socket.gethostname())
+
+    def generate_hash(self):
+        return hashlib.md5(str(time.time())).hexdigest()
+
+    def start_server(self):
+        self.s = socket.socket()
+        try:
+            self.s.bind((self.host, self.port))
+            logs.logging.info("XXE: Server started.")      
+            return True 
+        except socket.error:
+            logs.logging.info("XXE: Can't bind to port. Port may be busy or check firewall setting.")      
+
+    def start_listening(self):
+        try:
+            while True:
+                # Wait for 5 seconds
+                self.s.listen(5)                    
+                self.conn, self.addr = self.s.accept()
+                self.data = self.conn.recv(1024)
+                if self.data and unique_id in self.data:
+                    #External DTD is enable. URL is suspecious to XXE
+                    self.conn.sendall(data)
+                    global vulnerable
+                    vulnerable = True
+
+            self.conn.close()
+
+        except socket.error:
+            print "[-]URL might not be vulnerable to XXE. We reccomend you to check it manually"
+            self.conn.close()
+
+    def fetch_xxe_payload(self):
+        # Returns xxe payloads in list type
+        payload_list = []
+        if os.getcwd().split('/')[-1] == 'API':
+            path = '../Payloads/xxe.txt'
+        else:
+            path = 'Payloads/xxe.txt'
+
+        with open(path) as f:
+            for line in f:
+                if line:
+                    payload_list.append(line.rstrip())
+
+        return payload_list
+
+    def send_request(self,url,method,temp_headers,xxe_payloads,scanid=None):
+        # Test if if server is accepiting XML data
+        sample_xml = '''<?xml version="1.0" encoding="UTF-8"?><text>hello world</text>'''
+        xml_request = requests.post(url, headers=temp_headers, data=sample_xml)
+        if xml_request.status_code == 415:
+            # Media type not supported. 
+            return 
+        global unique_id
+        unique_id = self.generate_hash()
+        host = "http://"+str(self.host)+":"+str(self.port)+"/"+unique_id
+        for payload in xxe_payloads:
+            payload = payload.replace("{host}",host)
+            xxe_request = requests.post(url, headers=temp_headers, data=payload)
+            time.sleep(10)
+            if vulnerable is True:
+                print "[+]{0} is vulnerable to XML External Entity Attack".format(url)
+                attack_result = { "id" : 14, "scanid" : scanid, "url" : url, "alert": "XML External Entity Attack", "impact": "High", "req_headers": temp_headers, "req_body":payload, "res_headers": xxe_request.headers ,"res_body": xxe_request.text}
+                dbupdate.insert_record(attack_result)
+                break
+
+    def xxe_test(self,url,method,headers,body,scanid=None):
+        temp_headers = {}
+        temp_headers.update(headers)
+        xxe = xxe_scan()
+        socketresult = xxe.start_server()
+        if socketresult is True:
+            t = threading.Thread(target=xxe.start_listening)
+            t.daemon = True
+            t.start()
+            temp_headers['Content-Type'] = 'text/xml'
+            xxe_payloads = self.fetch_xxe_payload()
+            self.send_request(url,method,temp_headers,xxe_payloads,scanid)  

utils/scan.property

@@ -8,7 +8,8 @@ attack = {
           "jwt" : 'y',
           "sqli" : 'y',
           "xss" : 'y',
-          "open-redirection" : "y"
+          "open-redirection" : "y",
+          "xxe" : "y"
          }
 
 

utils/vulnerabilities.py

@@ -74,5 +74,11 @@
       'name': 'Open redirection',
       'Description':  'Unvalidated redirects and forwards are possible when an application accepts untrusted input that could cause an application to redirect the request to a URL contained within untrusted input. By modifying untrusted URL input to a malicious site, an attacker may successfully launch a phishing scam and steal user credentials. Because the server name in the modified link is identical to the original site, phishing attempts may have a more trustworthy appearance. Unvalidated redirect and forward attacks can also be used to maliciously craft a URL that would pass the application’s access control check and then forward the attacker to privileged functions that they would normally not be able to access.',
       'remediation': 'Sanitize input by creating a list of trusted URL\'s (lists of hosts or a regex).'
+      },
+      {
+      'id': 14,
+      'name': 'XML External Entity Attack',
+      'Description':  'An XML External Entity attack is a type of attack against an application that parses XML input. This attack occurs when XML input containing a reference to an external entity is processed by a weakly configured XML parser. This attack may lead to the disclosure of confidential data, denial of service, server side request forgery, port scanning from the perspective of the machine where the parser is located, and other system impacts.',
+      'remediation': 'The XML processor should be configured to use a local static DTD and disallow any declared DTD included in the XML document.'
       }
 ]