extra - cookies

aiopenapi3/extra/__init__.py

@@ -0,0 +1,4 @@
+from .reduce import Cull, Reduce
+from .cookies import Cookies
+
+__all__ = ["Cull", "Reduce", "Cookies"]

aiopenapi3/extra/cookies.py

@@ -0,0 +1,88 @@
+from typing import Literal
+import email.message
+import http.cookiejar
+import urllib.request
+
+import aiopenapi3.plugin
+
+
+class Cookies(aiopenapi3.plugin.Message, aiopenapi3.plugin.Init):
+    class _Request(urllib.request.Request):
+        """
+        c.f. httpx _CookieCompatRequest
+        """
+
+        def __init__(self, ctx: aiopenapi3.plugin.Message.Context) -> None:
+            super().__init__(
+                url=str(ctx.request.api.url),
+                headers=dict(ctx.headers),
+                method=ctx.request.method,
+            )
+            self.ctx = ctx
+
+        def add_unredirected_header(self, key: str, value: str) -> None:
+            assert key.lower() == "cookie"
+            name, _, value = value.partition("=")
+            self.ctx.cookies[name] = value
+
+    class _Response:
+        """
+        c.f. c.f. httpx _CookieCompatResponse
+        """
+
+        def __init__(self, ctx: aiopenapi3.plugin.Message.Context) -> None:
+            self.ctx = ctx
+
+        def info(self) -> email.message.Message:
+            info = email.message.Message()
+            key = "set-cookie"
+            for value in self.ctx.headers.get_list(key):
+                info[key] = value
+            return info
+
+    def __init__(
+        self, cookiejar: http.cookiejar.CookieJar = None, policy: Literal["jar", "securitySchemes"] = "jar"
+    ) -> None:
+        self.cookiejar: http.cookiejar.CookieJar = cookiejar or http.cookiejar.CookieJar()
+        self.policy: Literal["jar", "securitySchemes"] = policy
+        self.schemes: dict[str, str] = dict()
+
+        if policy not in ["jar", "securitySchemes"]:
+            raise ValueError(f"policy {self.policy} is not a valid policy")
+
+        super().__init__()
+
+    def initialized(self, ctx: "aiopenapi3.plugin.Init.Context") -> "aiopenapi3.plugin.Init.Context":
+        self.schemes = {
+            v.root.name: k
+            for k, v in filter(
+                lambda x: (x[1].root.type.lower(), x[1].root.in_) == ("apikey", "cookie"),
+                self.api.components.securitySchemes.items(),
+            )
+        }
+        return ctx
+
+    def received(self, ctx: "aiopenapi3.plugin.Message.Context") -> "aiopenapi3.plugin.Message.Context":
+        response = Cookies._Response(ctx)
+        request = Cookies._Request(ctx)
+
+        cookies = self.cookiejar.make_cookies(response, request)
+
+        for cookie in cookies:
+            if not self.cookiejar._policy.set_ok(cookie, request):
+                continue  # pragma: no cover
+
+            if (ss := self.schemes.get(cookie.name)) is not None:
+                self.api.authenticate(**{ss: cookie.value})
+            elif self.policy == "jar":
+                self.cookiejar.set_cookie(cookie)
+
+        return ctx
+
+    def sending(self, ctx: "aiopenapi3.plugin.Message.Context") -> "aiopenapi3.plugin.Message.Context":
+        if self.policy == "jar":
+            self.cookiejar.add_cookie_header(Cookies._Request(ctx))
+        elif self.policy == "securitySchemes":
+            # authentication will take care
+            pass
+        return ctx

aiopenapi3/extra.py → aiopenapi3/extra/reduce.py

@@ -3,10 +3,10 @@
 import logging
 import re
 
-from .plugin import Document, Init
+from ..plugin import Document, Init
 
 if typing.TYPE_CHECKING:
-    from ._types import HTTPMethodMatchType, PathItemType
+    from .._types import HTTPMethodMatchType, PathItemType
 
     PathMatchType = Union[re.Pattern, str]
     OperationIdMatchType = Union[re.Pattern, str]

aiopenapi3/plugin.py

@@ -17,7 +17,8 @@
     from aiopenapi3 import OpenAPI
 
     import httpx
-    from .base import PathItemBase, SchemaBase, RequestBase
+    from .base import PathItemBase, SchemaBase
+    from .request import RequestBase
 
 """
 the plugin interface replicates the suds way of  dealing with broken data/schema information
@@ -117,6 +118,8 @@ class Context:
         """available :func:`~aiopenapi3.plugin.Message.received` """
         headers: "httpx.Headers" = None
         """available :func:`~aiopenapi3.plugin.Message.sending` :func:`~aiopenapi3.plugin.Message.received` """
+        cookies: dict[str, str] = None
+        """available :func:`~aiopenapi3.plugin.Message.sending` """
         status_code: Optional[str] = None
         """available :func:`~aiopenapi3.plugin.Message.received` """
         content_type: Optional[str] = None

aiopenapi3/v30/glue.py

@@ -172,7 +172,7 @@ def _prepare_secschemes_default(self, scheme: str, value: Union[str, Sequence[st
                 self.req.headers[ss.name] = value
 
             if ss.in_ == "cookie":
-                self.req.cookies = {ss.name: value}
+                self.req.cookies[ss.name] = value
 
     def _prepare_secschemes_extra(self, scheme: str, value: Union[str, Sequence[str]]) -> None:
         assert (
@@ -251,7 +251,7 @@ def _prepare_secschemes_extra(self, scheme: str, value: Union[str, Sequence[str]
                 auths.append(auth(value, ss.name))
 
             if ss.in_ == "cookie":
-                self.req.cookies = {ss.name: value}
+                self.req.cookies[ss.name] = value
 
         for auth in auths:
             if self.req.auth and isinstance(self.req.auth, SupportMultiAuth):
@@ -341,10 +341,15 @@ def _prepare_body(self, data_: Optional["RequestData"], rbq: dict[str, str]) ->
 
         if not self.operation.requestBody:
             ctx = self.api.plugins.message.sending(
-                request=self, operationId=self.operation.operationId, sending=None, headers=self.req.headers
+                request=self,
+                operationId=self.operation.operationId,
+                sending=None,
+                headers=self.req.headers,
+                cookies=self.req.cookies,
             )
             self.req.content = ctx.sending
             self.req.headers = ctx.headers
+            self.req.cookies = ctx.cookies
             return
 
         if data_ is None and self.operation.requestBody.required:
@@ -364,10 +369,15 @@ def _prepare_body(self, data_: Optional["RequestData"], rbq: dict[str, str]) ->
             data: bytes = data.encode()  # type: ignore[union-attr]
             self.req.headers["Content-Type"] = "application/json"
             ctx = self.api.plugins.message.sending(
-                request=self, operationId=self.operation.operationId, sending=data, headers=self.req.headers
+                request=self,
+                operationId=self.operation.operationId,
+                sending=data,
+                headers=self.req.headers,
+                cookies=self.req.cookies,
             )
             self.req.content = ctx.sending
             self.req.headers = ctx.headers
+            self.req.cookies = ctx.cookies
         elif (ct := "multipart/form-data") in self.operation.requestBody.content:
             """
             https://swagger.io/docs/specification/describing-request-body/multipart-requests/
@@ -413,6 +423,18 @@ def _prepare_body(self, data_: Optional["RequestData"], rbq: dict[str, str]) ->
             else:
                 assert media.schema_
                 raise TypeError((type(data_), media.schema_.get_type()))
+
+            # sending is unset here
+            ctx = self.api.plugins.message.sending(
+                request=self,
+                operationId=self.operation.operationId,
+                sending=None,
+                headers=self.req.headers,
+                cookies=self.req.cookies,
+            )
+            self.req.headers = ctx.headers
+            self.req.cookies = ctx.cookies
+
         elif (ct := "application/x-www-form-urlencoded") in self.operation.requestBody.content:
             self.req.headers["Content-Type"] = ct
             media: aiopenapi3.v30.media.MediaType = self.operation.requestBody.content[ct]
@@ -424,6 +446,18 @@ def _prepare_body(self, data_: Optional["RequestData"], rbq: dict[str, str]) ->
             params = parameters_from_urlencoded(data_, media)
             content = urllib.parse.urlencode(params, doseq=True)
             self.req.content = content
+
+            ctx = self.api.plugins.message.sending(
+                request=self,
+                operationId=self.operation.operationId,
+                sending=self.req.content,
+                headers=self.req.headers,
+                cookies=self.req.cookies,
+            )
+            self.req.content = ctx.sending
+            self.req.headers = ctx.headers
+            self.req.cookies = ctx.cookies
+
         elif (ct := "application/octet-stream") in self.operation.requestBody.content:
             self.req.headers["Content-Type"] = ct
             value: "RequestFileParameter"
@@ -434,6 +468,18 @@ def _prepare_body(self, data_: Optional["RequestData"], rbq: dict[str, str]) ->
                 self.req.content = data_
             else:
                 raise TypeError(data_)
+
+            ctx = self.api.plugins.message.sending(
+                request=self,
+                operationId=self.operation.operationId,
+                sending=self.req.content,
+                headers=self.req.headers,
+                cookies=self.req.cookies,
+            )
+            self.req.content = ctx.sending
+            self.req.headers = ctx.headers
+            self.req.cookies = ctx.cookies
+
         else:
             raise NotImplementedError(self.operation.requestBody.content)
 

docs/source/api.rst

@@ -396,3 +396,24 @@ Code below will eleminate all schemas not required to serve the operations ident
 .. autoclass:: Reduce
     :members: __init__
 .. autoclass:: Cull
+
+
+Cookies
+-------
+
+This plugin deals with cookies from responses and adds in requests based on a policy.
+
+There are two policies:
+* securitySchemes - only cookies whose name is in the securitySchemes are used. To match authentication requirements credentials are set via OpenAPI.authenticate(name=value)
+* jar - all cookies are used, basically like a browser would do. Cookies not mentioned in securitySchemes are set besides OpenAPI.authenticate() to allow using them without adjusting the description document.
+
+.. code:: python
+
+    api = OpenAPI.load_sync(
+        "http://127.0.0.1/api.yaml",
+        plugins=[
+            Cookies(policy="jar")
+        ],
+    )
+
+.. autoclass:: Cookies

docs/source/extra.rst

@@ -18,3 +18,7 @@ Large description documents which are autogenerated by converting other service
 may benefit from additional changes to the description document to eliminate conversion artifacts depending on the converter used.
 
 As an example for additional steps based on the `Microsoft Graph API <https://github.com/microsoftgraph/msgraph-metadata>`_ refer to :aioai3:ref:`tests.extra_test.MSGraph`.
+
+Cookies
+=======
+To assist in dealing with cookie requirements of some APIs, :class:`aiopenapi3.extra.Cookies` can be used.

tests/conftest.py

@@ -380,6 +380,11 @@ def with_extra_reduced():
     yield "extra-reduced.yaml"
 
 
+@pytest.fixture
+def with_extra_cookie(openapi_version):
+    yield _get_parsed_yaml("extra-cookie.yaml", openapi_version)
+
+
 @pytest.fixture
 def with_schema_self_recursion(openapi_version):
     yield _get_parsed_yaml("schema-self-recursion.yaml", openapi_version)

tests/extra_test.py

@@ -199,3 +199,44 @@ def test_reduced(with_extra_reduced, httpx_mock, compressor):
     assert "A1" in api.components.schemas
     assert "A" in api.components.responses
     assert "A" in api.components.requestBodies
+
+
+from aiopenapi3.extra import Cookies
+
+
+@pytest.mark.parametrize("cookie", [dict(policy="jar"), dict(policy="securitySchemes")], ids=["jar", "securityScheme"])
+def test_cookies(httpx_mock, with_extra_cookie, cookie):
+
+    api = OpenAPI(
+        "http://127.0.0.1/api.yaml",
+        with_extra_cookie,
+        session_factory=httpx.Client,
+        plugins=[Cookies(**cookie)],
+    )
+
+    httpx_mock.add_response(
+        url="http://127.0.0.1/api/set-cookie",
+        headers=[("Set-Cookie", "Session=value"), ("Set-Cookie", "a=b")],
+        json='"ok"',
+    )
+    api._.set_cookie()
+
+    if cookie["policy"] == "jar":
+        httpx_mock.add_response(
+            url="http://127.0.0.1/api/require-cookie", match_headers={"Cookie": "Session=value; a=b"}, json='"ok"'
+        )
+    else:
+        httpx_mock.add_response(
+            url="http://127.0.0.1/api/require-cookie", match_headers={"Cookie": "Session=value"}, json='"ok"'
+        )
+    api._.require_cookie()
+
+    req = httpx_mock.get_requests()[-1]
+
+    if cookie["policy"] == "securitySchemes":
+        assert req.headers.get_list("cookie") == ["Session=value"]
+
+
+def test_cookies_policy(with_extra_cookie):
+    with pytest.raises(ValueError, match="policy … is not a valid policy"):
+        Cookies(policy="…")

tests/fixtures/extra-cookie.yaml

@@ -0,0 +1,41 @@
+openapi: 3.0.3
+info:
+  title: ''
+  version: 0.0.0
+servers:
+  - url: http://127.0.0.1/api
+
+security:
+  - cookieAuth: []
+
+paths:
+  /set-cookie:
+    get:
+      operationId: set_cookie
+      responses:
+        '200':
+          content:
+            application/json:
+              schema:
+                type: string
+          description: ''
+      security: []
+
+  /require-cookie:
+    get:
+      operationId: require_cookie
+      description: ''
+      responses:
+        '200':
+          content:
+            application/json:
+              schema:
+                type: string
+          description: ''
+
+components:
+  securitySchemes:
+    cookieAuth:
+      type: apiKey
+      in: cookie
+      name: Session