Add possibility to skip https ssl check for schemathesis

lobziik · Stranger6667 · commit 7f80e47316c3 · 2022-07-14T16:05:01.000+02:00
This commit adds new flag to the wafp cli -
`--fuzzer-skip-ssl-verify` which tells fuzzeers
to skip certs ferification.

Also added possibility to disable certs verification on per
target basis, for such purposes new field `fuzzer_skip_ssl_verify`
was introduced in target context. After target run its value
will be piped down to a fuzzer run function.
diff --git a/src/wafp/__main__.py b/src/wafp/__main__.py
@@ -18,6 +18,7 @@ class CliArguments(targets.cli.SharedCliArguments, fuzzers.cli.SharedCliArgument
 
     build: bool
     output_dir: str
+    fuzzer_skip_ssl_verify: bool
 
     @classmethod
     def from_all_args(
@@ -39,6 +40,13 @@ def extend_parser(cls, parser: argparse.ArgumentParser, *, catalog: Optional[str
         parser.add_argument(
             "--build", action="store_true", required=False, default=False, help="Force building docker images"
         )
+        parser.add_argument(
+            "--fuzzer-skip-ssl-verify",
+            action="store_true",
+            required=False,
+            default=False,
+            help="Tells fuzzer to skip certificates verification in case of https",
+        )
         parser.add_argument(
             "--output-dir",
             action="store",
@@ -65,6 +73,7 @@ def main(
             schema=context.schema_location,
             base_url=context.base_url,
             headers=context.headers,
+            ssl_insecure=cli_args.fuzzer_skip_ssl_verify or context.fuzzer_skip_ssl_verify,
             build=cli_args.build,
             target=cli_args.target,
         ) as result:
diff --git a/src/wafp/fuzzers/catalog/api_fuzzer/__init__.py b/src/wafp/fuzzers/catalog/api_fuzzer/__init__.py
@@ -8,8 +8,16 @@
 
 class Default(BaseFuzzer):
     def get_entrypoint_args(
-        self, context: FuzzerContext, schema: str, base_url: str, headers: Dict[str, str]
+        self,
+        context: FuzzerContext,
+        schema: str,
+        base_url: str,
+        headers: Dict[str, str],
+        ssl_insecure: bool = False,
     ) -> List[str]:
+        if ssl_insecure:
+            self.logger.warning("Explicit cert verification skip is not supported for this fuzzer yet")
+
         args = ["--basic_output=True"]
         if is_url(schema):
             args.append(f"--src_url={schema}")
diff --git a/src/wafp/fuzzers/catalog/cats/__init__.py b/src/wafp/fuzzers/catalog/cats/__init__.py
@@ -25,8 +25,16 @@ def get_container_output_directory(self) -> pathlib.Path:
         return pathlib.Path("/app/test-report/")
 
     def get_entrypoint_args(
-        self, context: FuzzerContext, schema: str, base_url: str, headers: Dict[str, str]
+        self,
+        context: FuzzerContext,
+        schema: str,
+        base_url: str,
+        headers: Dict[str, str],
+        ssl_insecure: bool = False,
     ) -> List[str]:
+        if ssl_insecure:
+            self.logger.warning("Explicit cert verification skip is not supported for this fuzzer yet")
+
         args = [f"--contract={schema}", f"--server={base_url}"]
         if headers:
             # Over-simplified YAML serialization only for this exact case
diff --git a/src/wafp/fuzzers/catalog/fuzz_lightyear/__init__.py b/src/wafp/fuzzers/catalog/fuzz_lightyear/__init__.py
@@ -7,8 +7,16 @@
 
 class Default(BaseFuzzer):
     def get_entrypoint_args(
-        self, context: FuzzerContext, schema: str, base_url: str, headers: Dict[str, str]
+        self,
+        context: FuzzerContext,
+        schema: str,
+        base_url: str,
+        headers: Dict[str, str],
+        ssl_insecure: bool = False,
     ) -> List[str]:
+        if ssl_insecure:
+            self.logger.warning("Explicit cert verification skip is not supported for this fuzzer yet")
+
         if not is_url(schema):
             # The `url` argument is ignored if we pass the `--schema` option
             args = [f"--schema={schema}", "http://0.0.0.0/any.yaml"]
diff --git a/src/wafp/fuzzers/catalog/fuzzy_swagger/__init__.py b/src/wafp/fuzzers/catalog/fuzzy_swagger/__init__.py
@@ -5,7 +5,15 @@
 
 class Default(BaseFuzzer):
     def get_entrypoint_args(
-        self, context: FuzzerContext, schema: str, base_url: str, headers: Dict[str, str]
+        self,
+        context: FuzzerContext,
+        schema: str,
+        base_url: str,
+        headers: Dict[str, str],
+        ssl_insecure: bool = False,
     ) -> List[str]:
+        if ssl_insecure:
+            self.logger.warning("Explicit cert verification skip is not supported for this fuzzer yet")
+
         # Fuzzy-Swagger does not support custom headers
         return [f"--swagger={schema}", f"--server={base_url}", "--verbose"]
diff --git a/src/wafp/fuzzers/catalog/got_swag/__init__.py b/src/wafp/fuzzers/catalog/got_swag/__init__.py
@@ -5,7 +5,15 @@
 
 class Default(BaseFuzzer):
     def get_entrypoint_args(
-        self, context: FuzzerContext, schema: str, base_url: str, headers: Dict[str, str]
+        self,
+        context: FuzzerContext,
+        schema: str,
+        base_url: str,
+        headers: Dict[str, str],
+        ssl_insecure: bool = False,
     ) -> List[str]:
+        if ssl_insecure:
+            self.logger.warning("Explicit cert verification skip is not supported for this fuzzer yet")
+
         # Custom headers are only supported as variables for their tests DSL
         return [schema, "-m"]
diff --git a/src/wafp/fuzzers/catalog/restler/__init__.py b/src/wafp/fuzzers/catalog/restler/__init__.py
@@ -26,8 +26,16 @@ def prepare_schema(self, context: FuzzerContext, schema: str) -> str:
         return str(container_input / filename)
 
     def get_entrypoint_args(
-        self, context: FuzzerContext, schema: str, base_url: str, headers: Dict[str, str]
+        self,
+        context: FuzzerContext,
+        schema: str,
+        base_url: str,
+        headers: Dict[str, str],
+        ssl_insecure: bool = False,
     ) -> List[str]:
+        if ssl_insecure:
+            self.logger.warning("Explicit cert verification skip is not supported for this fuzzer yet")
+
         parsed = urlparse(base_url)
         args = [
             str(self.get_container_output_directory()),
diff --git a/src/wafp/fuzzers/catalog/schemathesis/__init__.py b/src/wafp/fuzzers/catalog/schemathesis/__init__.py
@@ -7,7 +7,7 @@
 
 class Default(BaseFuzzer):
     def get_entrypoint_args(
-        self, context: FuzzerContext, schema: str, base_url: str, headers: Dict[str, str]
+        self, context: FuzzerContext, schema: str, base_url: str, headers: Dict[str, str], ssl_insecure: bool = False
     ) -> List[str]:
         args = [
             "run",
@@ -22,6 +22,8 @@ def get_entrypoint_args(
         if headers:
             for key, value in headers.items():
                 args.extend(["-H", f"{key}: {value}"])
+        if ssl_insecure:
+            args.extend(["--request-tls-verify=false"])
         extend_entrypoint_args(context, args)
         return args
 
@@ -36,27 +38,27 @@ def extend_entrypoint_args(context: FuzzerContext, args: List[str]) -> None:
 
 class AllChecks(Default):
     def get_entrypoint_args(
-        self, context: FuzzerContext, schema: str, base_url: str, headers: Dict[str, str]
+        self, context: FuzzerContext, schema: str, base_url: str, headers: Dict[str, str], ssl_insecure: bool = False
     ) -> List[str]:
-        args = super().get_entrypoint_args(context, schema, base_url, headers)
+        args = super().get_entrypoint_args(context, schema, base_url, headers, ssl_insecure)
         args.append("--checks=all")
         return args
 
 
 class Negative(Default):
     def get_entrypoint_args(
-        self, context: FuzzerContext, schema: str, base_url: str, headers: Dict[str, str]
+        self, context: FuzzerContext, schema: str, base_url: str, headers: Dict[str, str], ssl_insecure: bool = False
     ) -> List[str]:
-        args = super().get_entrypoint_args(context, schema, base_url, headers)
+        args = super().get_entrypoint_args(context, schema, base_url, headers, ssl_insecure)
         args.append("--data-generation-method=negative")
         return args
 
 
 class StatefulOld(Default):
     def get_entrypoint_args(
-        self, context: FuzzerContext, schema: str, base_url: str, headers: Dict[str, str]
+        self, context: FuzzerContext, schema: str, base_url: str, headers: Dict[str, str], ssl_insecure: bool = False
     ) -> List[str]:
-        args = super().get_entrypoint_args(context, schema, base_url, headers)
+        args = super().get_entrypoint_args(context, schema, base_url, headers, ssl_insecure)
         args.append("--stateful=links")
         return args
 
@@ -66,8 +68,11 @@ def get_entrypoint(self) -> Union[str, NotSet]:
         return "pytest"
 
     def get_entrypoint_args(
-        self, context: FuzzerContext, schema: str, base_url: str, headers: Dict[str, str]
+        self, context: FuzzerContext, schema: str, base_url: str, headers: Dict[str, str], ssl_insecure: bool = False
     ) -> List[str]:
+        if ssl_insecure:
+            self.logger.warning("Explicit cert verification skip is not supported for this target yet")
+
         filename = "test_stateful.py"
         if context.target is not None and context.target.startswith("age_of_empires_2_api"):
             extra = {"force_schema_version": "30"}
diff --git a/src/wafp/fuzzers/catalog/swagger_conformance/__init__.py b/src/wafp/fuzzers/catalog/swagger_conformance/__init__.py
@@ -5,7 +5,15 @@
 
 class Default(BaseFuzzer):
     def get_entrypoint_args(
-        self, context: FuzzerContext, schema: str, base_url: str, headers: Dict[str, str]
+        self,
+        context: FuzzerContext,
+        schema: str,
+        base_url: str,
+        headers: Dict[str, str],
+        ssl_insecure: bool = False,
     ) -> List[str]:
+        if ssl_insecure:
+            self.logger.warning("Explicit cert verification skip is not supported for this fuzzer yet")
+
         # Swagger-conformance does not support setting base URL or custom headers
         return [schema]
diff --git a/src/wafp/fuzzers/catalog/swagger_fuzzer/__init__.py b/src/wafp/fuzzers/catalog/swagger_fuzzer/__init__.py
@@ -12,7 +12,15 @@ def prepare_schema(self, context: FuzzerContext, schema: str) -> str:
         return self.serve_spec(context, schema)
 
     def get_entrypoint_args(
-        self, context: FuzzerContext, schema: str, base_url: str, headers: Dict[str, str]
+        self,
+        context: FuzzerContext,
+        schema: str,
+        base_url: str,
+        headers: Dict[str, str],
+        ssl_insecure: bool = False,
     ) -> List[str]:
+        if ssl_insecure:
+            self.logger.warning("Explicit cert verification skip is not supported for this fuzzer yet")
+
         # Swagger-fuzzer does not support setting base URL or custom headers
         return [schema]
diff --git a/src/wafp/fuzzers/catalog/tnt_fuzzer/__init__.py b/src/wafp/fuzzers/catalog/tnt_fuzzer/__init__.py
@@ -14,8 +14,16 @@ def prepare_schema(self, context: FuzzerContext, schema: str) -> str:
         return self.serve_spec(context, schema)
 
     def get_entrypoint_args(
-        self, context: FuzzerContext, schema: str, base_url: str, headers: Dict[str, str]
+        self,
+        context: FuzzerContext,
+        schema: str,
+        base_url: str,
+        headers: Dict[str, str],
+        ssl_insecure: bool = False,
     ) -> List[str]:
+        if ssl_insecure:
+            self.logger.warning("Explicit cert verification skip is not supported for this fuzzer yet")
+
         parsed = urlparse(base_url)
         args = [
             f"--url={schema}",
diff --git a/src/wafp/fuzzers/core.py b/src/wafp/fuzzers/core.py
@@ -74,6 +74,7 @@ def start(
         schema: str,
         base_url: str,
         headers: Optional[Dict[str, str]] = None,
+        ssl_insecure: bool = False,
         build: bool = False,
         target: Optional[str] = None,
     ) -> "FuzzResult":
@@ -90,7 +91,7 @@ def start(
         start = time.perf_counter()
         completed_process = self.compose.run(
             service=self.get_fuzzer_service_name(),
-            args=self.get_entrypoint_args(context, schema_location, base_url, headers),
+            args=self.get_entrypoint_args(context, schema_location, base_url, headers, ssl_insecure),
             entrypoint=self.get_entrypoint(),
             volumes=self.get_volumes(context),
         )
@@ -104,6 +105,7 @@ def run(
         schema: str,
         base_url: str,
         headers: Optional[Dict[str, str]] = None,
+        ssl_insecure: bool = False,
         build: bool = False,
         target: Optional[str] = None,
     ) -> Generator["FuzzResult", None, None]:
@@ -112,7 +114,13 @@ def run(
         It is a common workflow for CLI.
         """
         try:
-            yield self.start(schema, base_url, headers, build, target)
+            yield self.start(
+                schema,
+                base_url,
+                headers,
+                ssl_insecure,
+                build,
+            )
         except subprocess.CalledProcessError as exc:
             sys.exit(exc.returncode)
         finally:
@@ -130,7 +138,12 @@ def serve_spec(self, context: "FuzzerContext", schema: str) -> str:
 
     @abc.abstractmethod
     def get_entrypoint_args(
-        self, context: "FuzzerContext", schema: str, base_url: str, headers: Dict[str, str]
+        self,
+        context: "FuzzerContext",
+        schema: str,
+        base_url: str,
+        headers: Dict[str, str],
+        ssl_insecure: bool = False,
     ) -> List[str]:
         """Arguments to fuzzer's entrypoint in `docker-compose run <service> <args>`."""
         raise NotImplementedError
diff --git a/src/wafp/targets/core.py b/src/wafp/targets/core.py
@@ -26,6 +26,7 @@ def generate_run_id() -> str:
 class BaseTarget(abc.ABC, Component):
     port: int = attr.ib(factory=unused_port)
     force_build: bool = attr.ib(default=False)
+    fuzzer_skip_ssl_verify: bool = attr.ib(default=False)
     sentry_dsn: Optional[str] = attr.ib(default=None)
     run_id: str = attr.ib(factory=generate_run_id)
     wait_target_ready_timeout: int = WAIT_TARGET_READY_TIMEOUT
@@ -64,7 +65,12 @@ def start(self, extra_env: Optional[Dict[str, str]] = None) -> "TargetContext":
         if headers:
             info["headers"] = headers
         self.logger.msg("Target is ready", **info)
-        return TargetContext(base_url=base_url, schema_location=self.get_schema_location(), headers=headers)
+        return TargetContext(
+            base_url=base_url,
+            schema_location=self.get_schema_location(),
+            headers=headers,
+            fuzzer_skip_ssl_verify=self.fuzzer_skip_ssl_verify,
+        )
 
     # These methods are expected to be overridden
 
@@ -173,6 +179,7 @@ class TargetContext:
     base_url: str = attr.ib()
     schema_location: str = attr.ib()
     headers: Dict[str, str] = attr.ib()
+    fuzzer_skip_ssl_verify: bool = attr.ib(default=False)
 
 
 Target = Type[BaseTarget]
diff --git a/test/fuzzers/fuzzers_catalog/example_fuzzer/__init__.py b/test/fuzzers/fuzzers_catalog/example_fuzzer/__init__.py
@@ -5,7 +5,12 @@
 
 class Default(BaseFuzzer):
     def get_entrypoint_args(
-        self, context: FuzzerContext, schema: str, base_url: str, headers: Optional[Dict[str, str]]
+        self,
+        context: FuzzerContext,
+        schema: str,
+        base_url: str,
+        headers: Optional[Dict[str, str]],
+        ssl_insecure: bool = False,
     ) -> List[str]:
         args = [schema]
         if headers is not None:
