Adding --openapi-tags

Author: stamparm

lib/core/option.py

@@ -521,8 +521,13 @@ def _setOpenApiTargets():
         logger.info(infoMsg)
         content = openFile(conf.openApiFile).read()
 
+    tags = [_.strip() for _ in re.split(PARAMETER_SPLITTING_REGEX, conf.openApiTags) if _.strip()] if conf.openApiTags else None
+    if tags:
+        infoMsg = "restricting extraction to OpenAPI/Swagger operations tagged: %s" % ", ".join(tags)
+        logger.info(infoMsg)
+
     try:
-        targets = openApiTargets(content, origin)
+        targets = openApiTargets(content, origin, tags)
     except ValueError as ex:
         errMsg = "unable to parse the OpenAPI/Swagger specification ('%s')" % getSafeExString(ex)
         raise SqlmapSyntaxException(errMsg)

lib/core/optiondict.py

@@ -21,6 +21,7 @@
         "configFile": "string",
         "openApiFile": "string",
         "openApiBase": "string",
+        "openApiTags": "string",
     },
 
     "Request": {

lib/core/settings.py

@@ -20,7 +20,7 @@
 from thirdparty import six
 
 # sqlmap version (<major>.<minor>.<month>.<monthly commit>)
-VERSION = "1.10.7.57"
+VERSION = "1.10.7.58"
 TYPE = "dev" if VERSION.count('.') > 2 and VERSION.split('.')[-1] != '0' else "stable"
 TYPE_COLORS = {"dev": 33, "stable": 90, "pip": 34}
 VERSION_STRING = "sqlmap/%s#%s" % ('.'.join(VERSION.split('.')[:-1]) if VERSION.count('.') > 2 and VERSION.split('.')[-1] == '0' else VERSION, TYPE)

lib/parse/cmdline.py

@@ -150,6 +150,9 @@ def cmdLineParser(argv=None):
         target.add_argument("--openapi-base", dest="openApiBase",
             help="Base URL for a host-less OpenAPI/Swagger spec")
 
+        target.add_argument("--openapi-tags", dest="openApiTags",
+            help="Only derive targets from operations with these tag(s)")
+
         # Request options
         request = parser.add_argument_group("Request", "These options can be used to specify how to connect to the target URL")
 

lib/parse/openapi.py

@@ -206,15 +206,18 @@ def _baseUrl(spec, origin=None, servers=None):
 
 _METHODS = ("get", "post", "put", "delete", "patch", "options", "head")
 
-def openApiTargets(content, origin=None):
+def openApiTargets(content, origin=None, tags=None):
     """
     Returns a list of (url, method, data, headers) request tuples derived from an OpenAPI/Swagger
     specification. 'headers' is a list of (name, value) tuples (matching conf.httpHeaders). 'origin'
     (scheme://host[:port] of the specification's own location) is used only to resolve RELATIVE 'servers'
     entries - absolute server URLs are used as declared. Path parameters and header/cookie values carry
-    the custom injection mark so they become testable injection points.
+    the custom injection mark so they become testable injection points. 'tags' (list) restricts extraction
+    to operations declaring at least one of those OpenAPI tags (to scope a scan of a large API).
     """
 
+    tagSet = set(tags) if tags else None
+
     spec = _loadSpec(content)
     if not isinstance(spec, dict) or not isinstance(spec.get("paths"), dict) or not spec.get("paths"):
         errMsg = "no valid 'paths' object found in the provided OpenAPI/Swagger specification"
@@ -236,6 +239,8 @@ def openApiTargets(content, origin=None):
         for method, operation in item.items():
             if str(method).lower() not in _METHODS or not isinstance(operation, dict):   # str(): YAML keys can be non-string (e.g. 404, 'on'->bool)
                 continue
+            if tagSet is not None and not (tagSet & set(_ for _ in (operation.get("tags") or []) if isinstance(_, six.string_types))):
+                continue                                  # '--openapi-tags' filter: operation carries none of the requested tags
             try:
                 # effective base URL with OpenAPI precedence: operation 'servers' > path-item 'servers' > root
                 opServers = operation.get("servers") or item.get("servers")

tests/test_openapi.py

@@ -134,6 +134,25 @@ def test_header_and_cookie_params_are_injection_marked(self):
         self.assertEqual(headers["X-Api"], "k" + MARK)
         self.assertEqual(headers["Cookie"], "sess=v" + MARK)
 
+    def test_tag_filter_restricts_operations(self):
+        # '--openapi-tags' keeps only operations declaring at least one requested tag; untagged
+        # operations are dropped when a filter is active
+        spec = {"openapi": "3.0.0", "servers": [{"url": "https://api.test"}], "paths": {
+            "/users/{id}": {"get": {"tags": ["users"], "parameters": [{"name": "id", "in": "path", "schema": {"type": "integer"}}]}},
+            "/admin": {"post": {"tags": ["admin"], "parameters": [{"name": "q", "in": "query", "schema": {"type": "string"}}]}},
+            "/ping": {"get": {"parameters": [{"name": "x", "in": "query", "schema": {"type": "string"}}]}}}}
+        content = json.dumps(spec)
+
+        self.assertEqual(len(openApiTargets(content)), 3)                          # no filter -> everything
+        self.assertEqual(len(openApiTargets(content, tags=["nope"])), 0)          # no match -> nothing (incl. untagged)
+
+        users = openApiTargets(content, tags=["users"])
+        self.assertEqual(len(users), 1)
+        self.assertIn("/users/", users[0][0])
+
+        both = openApiTargets(content, tags=["users", "admin"])                    # union of tags
+        self.assertEqual(sorted(_[1] for _ in both), ["GET", "POST"])
+
     # --- graceful degradation: a broken/poorly-defined spec must never crash the parser ---
 
     def test_malformed_raises_valueerror(self):