First commit

Author: arxenix

.gitignore

@@ -0,0 +1,2 @@
+.idea/
+*.iml

README.md

@@ -0,0 +1,10 @@
+# Firebase-scanner
+
+This project contains various tools used for automated scanning and vulnerability discovery in firebase apps.
+
+
+# Components
+
+- `db-discovery.py` - This tool will aggregate various services (DNSDumpster, Shodan) to attempt to discover random firebase project codes.
+- `endpoint-discovery.py` - Run this tool on a wireshark .pcap or binary file to extract potential firebase db endpoints.
+- `scanner.py` - This tool will see what data and endpoints in the realtime DB are accessible (read/write info) and dump that information. It can also optionally dump everything that it can read. You can optionally give it an auth token and a list of endpoints (from endpt discovery script) to help it find more data.

db-discovery.py

@@ -0,0 +1,48 @@
+import sys
+
+import requests
+from argparse import ArgumentParser, FileType
+from dnsdumpster.DNSDumpsterAPI import DNSDumpsterAPI
+
+
+def dnsdumpster():
+    results = DNSDumpsterAPI().search('firebaseio.com')
+    return [domain['domain'] for domain in results['dns_records']['host']]
+
+
+def is_firebase_project(code: str) -> bool:
+    r = requests.get("https://{}.firebaseio.com".format(code))
+    return r.status_code != 404
+
+
+def has_realtime_db(code: str) -> bool:
+    r = requests.options("https://{}.firebaseio.com/.json".format(code))
+    return r.status_code != 423
+
+
+def discover_dbs(args):
+    db_candidates = []
+    if args.type == "dnsdumpster":
+        db_candidates = dnsdumpster()
+
+    print("Discovered DBs:")
+    for candidate in db_candidates:
+        if is_firebase_project(candidate) and has_realtime_db(candidate):
+            args.out.write(candidate + "\n")
+            if args.out != sys.stdout:
+                print(candidate)
+
+
+
+def parse_args():
+    parser = ArgumentParser()
+    parser.add_argument('type', help="Look for potential dbs through this specified method", choices=["dnsdumpster"])
+    parser.add_argument('--out', help="A file to dump results to", nargs='?', type=FileType('w'), default=sys.stdout)
+    args = parser.parse_args()
+    discover_dbs(args)
+
+
+
+
+if __name__ == '__main__':
+    parse_args()

endpoint-discovery.py

@@ -0,0 +1,72 @@
+from argparse import ArgumentParser, FileType
+import re
+import sys
+
+
+def search_pcap(args):
+    import pyshark
+    fname = args.file.name
+    args.file.close()
+    cap = pyshark.FileCapture(input_file=fname, display_filter="http.request.uri contains \"/firebaseio/\"")
+
+    endpoints = []
+    for packet in cap:
+        uri = packet.http.request.uri
+        pattern = re.compile("https?://(.+)\\.firebaseio\\.com/(.*)")
+        match = pattern.match(uri)
+        path = match.group(2)
+        endpoints.append(path)
+    return endpoints
+
+
+def search_binary_exact(args):
+    binary = args.file.read()
+    pattern = re.compile("https?://(.+)\\.firebaseio\\.com/(.*)")
+    matches = pattern.findall(binary)
+    endpoints = [match.group(2) for match in matches]
+    return endpoints
+
+
+def search_binary_strings(args):
+    binary = args.file.read()
+    pattern = re.compile("\\w{5,}")  # find all text with len >= 5
+    matches = pattern.findall(binary)
+    endpoints = [match.group(0) for match in matches]
+    return endpoints
+
+
+def discover_endpoints(args):
+    endpoints = []
+    if args.type == 'pcap':
+        endpoints = search_pcap(args)
+    elif args.type == 'binary_exact':
+        endpoints = search_binary_exact(args)
+    elif args.type == 'binary_strings':
+        endpoints = search_binary_strings(args)
+
+    cleaned_endpoints = []
+    for endpoint in endpoints:
+        if endpoint.startswith("/"):
+            endpoint = endpoints[1:]
+        if endpoint.endswith(".json"):
+            endpoint = endpoints[:-5]
+        cleaned_endpoints.append(endpoint)
+
+    endpoints = cleaned_endpoints
+    print("{} potential endpoints found.".format(len(endpoints)))
+    data = "\n".join(endpoints) + "\n"
+    args.out.write(data)
+    args.out.close()
+
+
+def parse_args():
+    parser = ArgumentParser()
+    parser.add_argument('type', help="Look for potential endpoints in the specified file type", choices=["pcap", "binary_exact", "binary_strings"])
+    parser.add_argument('file', help="The file to search through", type=FileType('w'))
+    parser.add_argument('--out', help="A file to dump results to", nargs='?', type=FileType('w'), default=sys.stdout)
+    args = parser.parse_args()
+    discover_endpoints(args)
+
+
+if __name__ == '__main__':
+    parse_args()

endpoints.txt

@@ -0,0 +1,5 @@
+test1
+test2
+test3
+test4
+

requirements.txt

@@ -0,0 +1,2 @@
+pypcapkit==0.14.3
+requests==2.21.0

scanner.py

@@ -0,0 +1,103 @@
+from typing import Tuple
+from io import StringIO
+from argparse import ArgumentParser, FileType
+import requests
+import json
+import sys
+
+common_endpoints = [
+	"",
+	"users",
+	# TODO - users/$uid - extract uid from JWT?
+	"groups",
+	"messages",
+	"posts",
+	"chats",
+]
+
+# TODO - attempt using different values to check value filters
+value_attempts = [
+	"1",
+	"0",
+	"-1",
+	"1566889891"
+	"true",
+	"false",
+	"a",
+	"null",
+	"-JSOpn9ZC54A4P4RoqVa",
+	"591dd66c-ffb0-4f7c-80f2-13345066a159"
+]
+
+
+def try_endpoint(args, endpoint: str) -> Tuple[bool, bool, str]:
+	url = 'https://{}.firebaseio.com/{}.json'.format(args.code, endpoint)
+	if args.auth is not None:
+		url += '?auth={}'.format(args.auth)
+
+	read_r = requests.get(url)
+	# POST nothing to test write access to avoid overwriting data
+	write_r = requests.post(url, data="null", headers={'Content-type': 'application/json'})
+	data = read_r.json() if read_r.status_code != 401 else None
+	print(data)
+	return read_r.status_code != 401, write_r.status_code != 401, data
+
+
+# from https://stackoverflow.com/a/7205107/2683545
+def merge(a, b, path=None):
+	"merges b into a"
+	if path is None: path = []
+	for key in b:
+		if key in a:
+			if isinstance(a[key], dict) and isinstance(b[key], dict):
+				merge(a[key], b[key], path + [str(key)])
+			elif a[key] == b[key]:
+				pass # same leaf value
+			else:
+				raise Exception('Conflict at %s' % '.'.join(path + [str(key)]))
+		else:
+			a[key] = b[key]
+	return a
+
+
+def scan_sites(args):
+	firebase_sites = args.sites.read().splitlines()
+	output = {}
+	for site in firebase_sites:
+		print("Scanning site {}...".format(site))
+		arg_endpoints = args.endpoints.read().splitlines()
+		endpoint_list = common_endpoints + arg_endpoints
+		endpoint_info = {}
+		db_dump = {}
+		for endpoint in endpoint_list:
+			dump_path = endpoint.split("/")
+			read_success, write_success, data = try_endpoint(args, endpoint)
+			if data is not None:
+				for part in dump_path:
+					if part not in db_dump:
+						db_dump[part] = {}
+					if isinstance(data, dict):
+						db_dump[part] = merge(db_dump[part], data)
+					else:
+						db_dump[part] = data
+			if read_success or write_success:
+				endpoint_info[endpoint] = {"read": read_success, "write": write_success}
+		output[site] = {"info": endpoint_info, "dump": db_dump}
+	args.out.write(json.dumps(output))
+	args.out.close()
+
+
+def parse_args():
+	parser = ArgumentParser()
+	parser.add_argument('sites', help="File containing list of firebase sites to scan [code].firebaseio.com", type=FileType('r'))
+	# parser.add_argument('--methods', help="A list of HTTP methods to try", nargs="*", default=firebase_http_methods, choices=firebase_http_methods)
+	parser.add_argument('--auth', help="An optional auth token to use")
+	parser.add_argument('--endpoints', help="A list of known endpoints to check", nargs='?', type=FileType('r'), default=StringIO(""))
+	parser.add_argument('--dirty', help="Should we modify the db to find more writable locations?", action="store_true", default=False)
+	parser.add_argument('--out', help="A file to dump info to", nargs='?', type=FileType('w'), default=sys.stdout)
+	args = parser.parse_args()
+	scan_sites(args)
+
+
+if __name__ == '__main__':
+	parse_args()