auth improvements: warn if not known (#60)

* clean up parser a bit

* fix ci settings again

* remove unneeded security scheme wrapper

* clean up security schemes a bit

* move security scheme detection into detector

* add warning for unsupported auth

Author: sh-rp

.github/workflows/test.yml

@@ -5,9 +5,11 @@ on:
   push:
     branches:
       - master
+      - devel
   pull_request:
     branches:
       - master
+      - devel
   workflow_dispatch:
 
 concurrency:

dlt_openapi/__init__.py

@@ -5,6 +5,8 @@
 from pathlib import Path
 from typing import Optional, cast
 
+import httpcore
+import httpx
 from loguru import logger
 
 from dlt_openapi.utils.misc import import_class_from_string
@@ -31,18 +33,20 @@ class Project:  # pylint: disable=too-many-instance-attributes
     def __init__(
         self,
         *,
+        doc: bytes,
         openapi: OpenapiParser,
         detector: BaseDetector,
         renderer: BaseRenderer,
         config: Config,
     ) -> None:
+        self.doc = doc
         self.openapi = openapi
         self.detector = detector
         self.renderer = renderer
         self.config = config
 
     def parse(self) -> None:
-        self.openapi.parse()
+        self.openapi.parse(self.doc)
 
     def detect(self) -> None:
         logger.info("Running heuristics on parsed output")
@@ -80,16 +84,37 @@ def print_warnings(self) -> None:
                     logger.warning(w.msg)
 
 
+def _get_document(*, url: Optional[str] = None, path: Optional[Path] = None, timeout: int = 60) -> bytes:
+    if url is not None and path is not None:
+        raise ValueError("Provide URL or Path, not both.")
+    if url is not None:
+        logger.info(f"Downloading spec from {url}")
+        try:
+            response = httpx.get(url, timeout=timeout)
+            logger.success("Download complete")
+            return response.content
+        except (httpx.HTTPError, httpcore.NetworkError) as e:
+            raise ValueError("Could not get OpenAPI document from provided URL") from e
+    elif path is not None:
+        logger.info(f"Reading spec from {path}")
+        return Path(path).read_bytes()
+    else:
+        raise ValueError("No URL or Path provided")
+
+
 def _get_project_for_url_or_path(  # pylint: disable=too-many-arguments
     url: Optional[str],
     path: Optional[Path],
     config: Config = None,
 ) -> Project:
+    doc = _get_document(url=url, path=path)
+
     renderer_cls = cast(BaseRenderer, import_class_from_string(config.renderer_class))
     detector_cls = cast(BaseDetector, import_class_from_string(config.detector_class))
 
     return Project(
-        openapi=OpenapiParser(config, url or path),
+        doc=doc,
+        openapi=OpenapiParser(config),
         detector=detector_cls(config),  # type: ignore
         renderer=renderer_cls(config),  # type: ignore
         config=config,

dlt_openapi/detector/default/__init__.py

@@ -42,6 +42,7 @@
     DataResponseUndetectedWarning,
     PrimaryKeyNotFoundWarning,
     UnresolvedPathParametersWarning,
+    UnsupportedSecuritySchemeWarning,
 )
 
 Tree = Dict[str, Union["str", "Tree"]]
@@ -58,6 +59,9 @@ def run(self, open_api: OpenapiParser) -> None:
         """Run the detector"""
         self.warnings = {}
 
+        # detect security stuff
+        self.detect_security_schemes(open_api)
+
         # discover stuff from responses
         self.detect_paginators_and_responses(open_api.endpoints)
 
@@ -78,6 +82,40 @@ def run(self, open_api: OpenapiParser) -> None:
             if params := e.unresolvable_path_param_names:
                 self._add_warning(UnresolvedPathParametersWarning(params), e)
 
+    def detect_security_schemes(self, open_api: OpenapiParser) -> None:
+        schemes = list(open_api.security_schemes.values())
+
+        # detect scheme settings
+        for scheme in schemes:
+
+            if scheme.type == "apiKey":
+                scheme.detected_secret_name = "api_key"
+                scheme.detected_auth_vars = f"""
+            "type": "api_key",
+            "api_key": api_key,
+            "name": "{scheme.name}",
+            "location": "{scheme.location}"
+"""
+            elif scheme.type == "http" and scheme.scheme == "basic":
+                scheme.detected_secret_name = "password"
+                scheme.detected_auth_vars = """
+            "type": "http_basic",
+            "username": "username",
+            "password": password,
+"""
+            elif scheme.type == "http" and scheme.scheme == "bearer":
+                scheme.detected_secret_name = "token"
+                scheme.detected_auth_vars = """
+            "type": "bearer",
+            "token": token,
+"""
+
+        # find default scheme
+        if len(schemes) and schemes[0].supported:
+            open_api.detected_default_security_scheme = schemes[0]
+        elif len(schemes) and not schemes[0].supported:
+            self._add_warning(UnsupportedSecuritySchemeWarning(schemes[0].name))
+
     def detect_resource_names(self, endpoints: EndpointCollection) -> None:
         """iterate all endpoints and find a strategy to select the right resource name"""
 

dlt_openapi/detector/default/warnings.py

@@ -26,3 +26,12 @@ class DataResponseNoBodyWarning(BaseDetectionWarning):
         "No json response schema defined on main data response. "
         + "Will not be able to detect primary key and some paginators."
     )
+
+
+class UnsupportedSecuritySchemeWarning(BaseDetectionWarning):
+    def __init__(self, security_scheme: str) -> None:
+        self.security_scheme = security_scheme
+        self.msg = (
+            f"Security Scheme {security_scheme} is not supported natively at this time. "
+            + "Please provide a custom implementation."
+        )

dlt_openapi/parser/context.py

@@ -1,12 +1,10 @@
-from dataclasses import dataclass
 from typing import Any, Dict, Optional, Tuple, Union
 
 import openapi_schema_pydantic as osp
 import referencing
 import referencing.jsonschema
 
 from dlt_openapi.parser.config import Config
-from dlt_openapi.utils.misc import ClassName
 
 TComponentClass = Union[
     osp.Schema,
@@ -20,41 +18,19 @@
 ]
 
 
-@dataclass
-class SecurityScheme:
-    os_security_scheme: osp.SecurityScheme
-    class_name: ClassName
-
-    @property
-    def type(self) -> str:
-        return self.os_security_scheme.type
-
-    @property
-    def scheme(self) -> str:
-        return self.os_security_scheme.scheme
-
-    @property
-    def name(self) -> str:
-        return self.os_security_scheme.name
-
-    @property
-    def location(self) -> str:
-        return self.os_security_scheme.security_scheme_in
-
-
 class OpenapiContext:
     spec: osp.OpenAPI
     spec_raw: Dict[str, Any]
+    config: Config
 
-    _component_cache: Dict[str, Dict[str, Any]]
-    security_schemes: Dict[str, SecurityScheme]
+    _component_cache: Dict[str, Dict[str, Any]] = {}
 
     def __init__(self, config: Config, spec: osp.OpenAPI, spec_raw: Dict[str, Any]) -> None:
         self.config = config
         self.spec = spec
         self.spec_raw = spec_raw
-        self._component_cache = {}
-        self.security_schemes = {}
+
+        # setup ref resolver
         resource = referencing.Resource(  # type: ignore[var-annotated, call-arg]
             contents=self.spec_raw, specification=referencing.jsonschema.DRAFT202012
         )
@@ -99,8 +75,3 @@ def parameter_from_reference(self, ref: Union[osp.Reference, osp.Parameter]) ->
         if isinstance(ref, osp.Parameter):
             return ref
         return osp.Parameter.parse_obj(self._component_from_reference(ref))
-
-    def get_security_scheme(self, name: str) -> SecurityScheme:
-        if name in self.security_schemes:
-            return self.security_schemes[name]
-        return None