26class SecureCookiesMiddleware(BaseHTTPMiddleware):
 27    """
 28    Provides automatic secure cookie support through symmetric Fernet encryption. As per
 29    the `cryptography` docs, security is achieved through:
 30
 31    - 128-bit key AES in CBC mode using PKCS7 padding for encryption.
 32    - HMAC using SHA256 for authentication.
 33    """
 34
 35    secrets: List[str]
 36    """
 37    A list of secrets to use when encrypting and decrypting secure cookies. Supplying
 38    multiple secrets enables the use of key rotation, whereby the first key is used for
 39    all encryption, including to re-encrypt cookies previously encrypted with any of
 40    the other list entries.
 41    """
 42
 43    def __init__(
 44        self,
 45        app: ASGIApp,
 46        secrets: List[str],
 47        discard_invalid: bool = True,
 48        cookie_path: Optional[str] = None,
 49        cookie_domain: Optional[str] = None,
 50        cookie_secure: Optional[bool] = None,
 51        cookie_httponly: Optional[bool] = None,
 52        cookie_samesite: Optional[Literal["strict", "lax", "none"]] = None,
 53        excluded_cookies: Optional[List[str]] = None,
 54        included_cookies: Optional[List[str]] = None,
 55    ) -> None:
 56        super().__init__(app)
 57        self.mfernet = MultiFernet(Fernet(s) for s in secrets)
 58        self.discard_invalid = discard_invalid
 59        """ Discard any invalid / unencrypted cookies present in the request."""
 60        self.cookie_path = cookie_path
 61        """ The Path field value to override on every cookie."""
 62        self.cookie_domain = cookie_domain
 63        """ The Domain field value to override on every cookie."""
 64        self.cookie_secure = cookie_secure
 65        """ The Secure field value to override on every cookie."""
 66        self.cookie_httponly = cookie_httponly
 67        """ The HttpOnly field value to override on every cookie."""
 68        self.cookie_samesite = cookie_samesite
 69        """ The SameStime field value to override on every cookie."""
 70        self.excluded_cookies = excluded_cookies
 71        """
 72        The list of cookies to ignore when decrypting and encrypting in
 73        the request --> response cycle. This option is mutually exclusive with
 74        `included_cookies`.
 75        """
 76        self.included_cookies = included_cookies
 77        """
 78        The list of cookies to decrypt and encrypt in the request --> response cycle.
 79        This option is mutually exclusive with `excluded_cookies`.
 80        """
 81
 82        if excluded_cookies and included_cookies:
 83            raise BadArgumentError(
 84                "It doesn't make sense to specify both excluded and included cookies."
 85                " Supply either `excluded_cookies` or `included_cookies` to restrict"
 86                " which cookies should be secure."
 87            )
 88
 89        if cookie_samesite:
 90            samesite = cookie_samesite.lower()
 91
 92            if samesite not in ["strict", "lax", "none"]:
 93                raise BadArgumentError(
 94                    "SameSite attribute must be either 'strict', 'lax' or 'none'"
 95                )
 96            elif samesite == "none" and not self.cookie_secure:
 97                warnings.warn(
 98                    "Insecure cookies with a SameSite='None' attribute may be rejected"
 99                    " on newer browser versions (draft-ietf-httpbis-rfc6265bis). See"
100                    " https://caniuse.com/same-site-cookie-attribute for compat notes."
101                )
102
103    def set_header(self, request: Request, header: str, value: str) -> None:
104        """
105        Adds the given Header to the available Request headers. If the Header already
106        exists, its value is overwritten.
107        """
108        hkey = header.encode("latin-1")
109        request.scope["headers"] = [
110            *(h for h in request.scope["headers"] if h[0] != hkey),
111            (hkey, value.encode("latin-1")),
112        ]
113
114    def decrypt(self, value: str) -> str:
115        """Decrypt the given value using any of the configured secrets."""
116        return self.mfernet.decrypt(value.encode()).decode()
117
118    def encrypt(self, value: str) -> str:
119        """Encrypt the given value using the first configured secret."""
120        return self.mfernet.encrypt(value.encode()).decode()
121
122    def should_process_cookie(self, cookie: str) -> bool:
123        """Determines if the cookie should be included for processing."""
124        return (
125            (not self.included_cookies and not self.excluded_cookies)
126            or (self.included_cookies is not None and cookie in self.included_cookies)
127            or (
128                self.excluded_cookies is not None
129                and cookie not in self.excluded_cookies
130            )
131        )
132
133    async def dispatch(
134        self, request: Request, call_next: RequestResponseEndpoint
135    ) -> Response:
136        if len(request.cookies):
137            cookies: SimpleCookie[str] = SimpleCookie()
138            for cookie, value in request.cookies.items():
139                if self.should_process_cookie(cookie):
140                    try:
141                        # try to decrypt the cookie and pass it along
142                        cookies[cookie] = self.decrypt(value)
143                    except InvalidToken:
144                        # delete invalid or unencrypted cookies unless disabled
145                        if not self.discard_invalid:
146                            cookies[cookie] = value
147
148            # serialize and set the decrypted cookies to the asgi scope headers
149            # we have to modify the scope directly because request objects are not
150            # passed between ASGI applications / middleware.
151            self.set_header(
152                request, "cookie", cookies.output(header="", sep=";").strip()
153            )
154
155        # propagate the modified request
156        response: Response = await call_next(request)
157
158        # Extract the cookie headers to be mutated
159        cookie_headers = response.headers.getlist("set-cookie")
160        del response.headers["set-cookie"]
161
162        for cookie_header in cookie_headers:
163            ncookie: SimpleCookie[str] = SimpleCookie(cookie_header)
164            key = next(iter(ncookie.keys()))
165
166            if self.should_process_cookie(key):
167                ncookie[key] = self.encrypt(ncookie[key].value)
168
169                # Mutate the cookie based on middleware defaults (if provided)
170                if self.cookie_path is not None:
171                    ncookie[key]["path"] = self.cookie_path
172
173                if self.cookie_domain is not None:
174                    ncookie[key]["domain"] = self.cookie_domain
175
176                if self.cookie_secure is not None:
177                    ncookie[key]["secure"] = self.cookie_secure
178
179                if self.cookie_httponly is not None:
180                    ncookie[key]["httponly"] = self.cookie_httponly
181
182                if self.cookie_samesite is not None:
183                    ncookie[key]["samesite"] = self.cookie_samesite
184
185            response.headers.append(
186                "set-cookie", ncookie.output(header="", sep=";").strip()
187            )
188
189        return response

19class BadArgumentError(Exception):
20    """
21    Generic error arguments with invalid values, or combinations of conflicting
22    arguments.
23    """