api_key.py 9.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320
  1. from typing import Annotated
  2. from annotated_doc import Doc
  3. from fastapi.openapi.models import APIKey, APIKeyIn
  4. from fastapi.security.base import SecurityBase
  5. from starlette.exceptions import HTTPException
  6. from starlette.requests import Request
  7. from starlette.status import HTTP_401_UNAUTHORIZED
  8. class APIKeyBase(SecurityBase):
  9. model: APIKey
  10. def __init__(
  11. self,
  12. location: APIKeyIn,
  13. name: str,
  14. description: str | None,
  15. scheme_name: str | None,
  16. auto_error: bool,
  17. ):
  18. self.auto_error = auto_error
  19. self.model: APIKey = APIKey(
  20. **{"in": location}, # ty: ignore[invalid-argument-type]
  21. name=name,
  22. description=description,
  23. )
  24. self.scheme_name = scheme_name or self.__class__.__name__
  25. def make_not_authenticated_error(self) -> HTTPException:
  26. """
  27. The WWW-Authenticate header is not standardized for API Key authentication but
  28. the HTTP specification requires that an error of 401 "Unauthorized" must
  29. include a WWW-Authenticate header.
  30. Ref: https://datatracker.ietf.org/doc/html/rfc9110#name-401-unauthorized
  31. For this, this method sends a custom challenge `APIKey`.
  32. """
  33. return HTTPException(
  34. status_code=HTTP_401_UNAUTHORIZED,
  35. detail="Not authenticated",
  36. headers={"WWW-Authenticate": "APIKey"},
  37. )
  38. def check_api_key(self, api_key: str | None) -> str | None:
  39. if not api_key:
  40. if self.auto_error:
  41. raise self.make_not_authenticated_error()
  42. return None
  43. return api_key
  44. class APIKeyQuery(APIKeyBase):
  45. """
  46. API key authentication using a query parameter.
  47. This defines the name of the query parameter that should be provided in the request
  48. with the API key and integrates that into the OpenAPI documentation. It extracts
  49. the key value sent in the query parameter automatically and provides it as the
  50. dependency result. But it doesn't define how to send that API key to the client.
  51. ## Usage
  52. Create an instance object and use that object as the dependency in `Depends()`.
  53. The dependency result will be a string containing the key value.
  54. ## Example
  55. ```python
  56. from fastapi import Depends, FastAPI
  57. from fastapi.security import APIKeyQuery
  58. app = FastAPI()
  59. query_scheme = APIKeyQuery(name="api_key")
  60. @app.get("/items/")
  61. async def read_items(api_key: str = Depends(query_scheme)):
  62. return {"api_key": api_key}
  63. ```
  64. """
  65. def __init__(
  66. self,
  67. *,
  68. name: Annotated[
  69. str,
  70. Doc("Query parameter name."),
  71. ],
  72. scheme_name: Annotated[
  73. str | None,
  74. Doc(
  75. """
  76. Security scheme name.
  77. It will be included in the generated OpenAPI (e.g. visible at `/docs`).
  78. """
  79. ),
  80. ] = None,
  81. description: Annotated[
  82. str | None,
  83. Doc(
  84. """
  85. Security scheme description.
  86. It will be included in the generated OpenAPI (e.g. visible at `/docs`).
  87. """
  88. ),
  89. ] = None,
  90. auto_error: Annotated[
  91. bool,
  92. Doc(
  93. """
  94. By default, if the query parameter is not provided, `APIKeyQuery` will
  95. automatically cancel the request and send the client an error.
  96. If `auto_error` is set to `False`, when the query parameter is not
  97. available, instead of erroring out, the dependency result will be
  98. `None`.
  99. This is useful when you want to have optional authentication.
  100. It is also useful when you want to have authentication that can be
  101. provided in one of multiple optional ways (for example, in a query
  102. parameter or in an HTTP Bearer token).
  103. """
  104. ),
  105. ] = True,
  106. ):
  107. super().__init__(
  108. location=APIKeyIn.query,
  109. name=name,
  110. scheme_name=scheme_name,
  111. description=description,
  112. auto_error=auto_error,
  113. )
  114. async def __call__(self, request: Request) -> str | None:
  115. api_key = request.query_params.get(self.model.name)
  116. return self.check_api_key(api_key)
  117. class APIKeyHeader(APIKeyBase):
  118. """
  119. API key authentication using a header.
  120. This defines the name of the header that should be provided in the request with
  121. the API key and integrates that into the OpenAPI documentation. It extracts
  122. the key value sent in the header automatically and provides it as the dependency
  123. result. But it doesn't define how to send that key to the client.
  124. ## Usage
  125. Create an instance object and use that object as the dependency in `Depends()`.
  126. The dependency result will be a string containing the key value.
  127. ## Example
  128. ```python
  129. from fastapi import Depends, FastAPI
  130. from fastapi.security import APIKeyHeader
  131. app = FastAPI()
  132. header_scheme = APIKeyHeader(name="x-key")
  133. @app.get("/items/")
  134. async def read_items(key: str = Depends(header_scheme)):
  135. return {"key": key}
  136. ```
  137. """
  138. def __init__(
  139. self,
  140. *,
  141. name: Annotated[str, Doc("Header name.")],
  142. scheme_name: Annotated[
  143. str | None,
  144. Doc(
  145. """
  146. Security scheme name.
  147. It will be included in the generated OpenAPI (e.g. visible at `/docs`).
  148. """
  149. ),
  150. ] = None,
  151. description: Annotated[
  152. str | None,
  153. Doc(
  154. """
  155. Security scheme description.
  156. It will be included in the generated OpenAPI (e.g. visible at `/docs`).
  157. """
  158. ),
  159. ] = None,
  160. auto_error: Annotated[
  161. bool,
  162. Doc(
  163. """
  164. By default, if the header is not provided, `APIKeyHeader` will
  165. automatically cancel the request and send the client an error.
  166. If `auto_error` is set to `False`, when the header is not available,
  167. instead of erroring out, the dependency result will be `None`.
  168. This is useful when you want to have optional authentication.
  169. It is also useful when you want to have authentication that can be
  170. provided in one of multiple optional ways (for example, in a header or
  171. in an HTTP Bearer token).
  172. """
  173. ),
  174. ] = True,
  175. ):
  176. super().__init__(
  177. location=APIKeyIn.header,
  178. name=name,
  179. scheme_name=scheme_name,
  180. description=description,
  181. auto_error=auto_error,
  182. )
  183. async def __call__(self, request: Request) -> str | None:
  184. api_key = request.headers.get(self.model.name)
  185. return self.check_api_key(api_key)
  186. class APIKeyCookie(APIKeyBase):
  187. """
  188. API key authentication using a cookie.
  189. This defines the name of the cookie that should be provided in the request with
  190. the API key and integrates that into the OpenAPI documentation. It extracts
  191. the key value sent in the cookie automatically and provides it as the dependency
  192. result. But it doesn't define how to set that cookie.
  193. ## Usage
  194. Create an instance object and use that object as the dependency in `Depends()`.
  195. The dependency result will be a string containing the key value.
  196. ## Example
  197. ```python
  198. from fastapi import Depends, FastAPI
  199. from fastapi.security import APIKeyCookie
  200. app = FastAPI()
  201. cookie_scheme = APIKeyCookie(name="session")
  202. @app.get("/items/")
  203. async def read_items(session: str = Depends(cookie_scheme)):
  204. return {"session": session}
  205. ```
  206. """
  207. def __init__(
  208. self,
  209. *,
  210. name: Annotated[str, Doc("Cookie name.")],
  211. scheme_name: Annotated[
  212. str | None,
  213. Doc(
  214. """
  215. Security scheme name.
  216. It will be included in the generated OpenAPI (e.g. visible at `/docs`).
  217. """
  218. ),
  219. ] = None,
  220. description: Annotated[
  221. str | None,
  222. Doc(
  223. """
  224. Security scheme description.
  225. It will be included in the generated OpenAPI (e.g. visible at `/docs`).
  226. """
  227. ),
  228. ] = None,
  229. auto_error: Annotated[
  230. bool,
  231. Doc(
  232. """
  233. By default, if the cookie is not provided, `APIKeyCookie` will
  234. automatically cancel the request and send the client an error.
  235. If `auto_error` is set to `False`, when the cookie is not available,
  236. instead of erroring out, the dependency result will be `None`.
  237. This is useful when you want to have optional authentication.
  238. It is also useful when you want to have authentication that can be
  239. provided in one of multiple optional ways (for example, in a cookie or
  240. in an HTTP Bearer token).
  241. """
  242. ),
  243. ] = True,
  244. ):
  245. super().__init__(
  246. location=APIKeyIn.cookie,
  247. name=name,
  248. scheme_name=scheme_name,
  249. description=description,
  250. auto_error=auto_error,
  251. )
  252. async def __call__(self, request: Request) -> str | None:
  253. api_key = request.cookies.get(self.model.name)
  254. return self.check_api_key(api_key)