http.py 13 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417
  1. import binascii
  2. from base64 import b64decode
  3. from typing import Annotated
  4. from annotated_doc import Doc
  5. from fastapi.exceptions import HTTPException
  6. from fastapi.openapi.models import HTTPBase as HTTPBaseModel
  7. from fastapi.openapi.models import HTTPBearer as HTTPBearerModel
  8. from fastapi.security.base import SecurityBase
  9. from fastapi.security.utils import get_authorization_scheme_param
  10. from pydantic import BaseModel
  11. from starlette.requests import Request
  12. from starlette.status import HTTP_401_UNAUTHORIZED
  13. class HTTPBasicCredentials(BaseModel):
  14. """
  15. The HTTP Basic credentials given as the result of using `HTTPBasic` in a
  16. dependency.
  17. Read more about it in the
  18. [FastAPI docs for HTTP Basic Auth](https://fastapi.tiangolo.com/advanced/security/http-basic-auth/).
  19. """
  20. username: Annotated[str, Doc("The HTTP Basic username.")]
  21. password: Annotated[str, Doc("The HTTP Basic password.")]
  22. class HTTPAuthorizationCredentials(BaseModel):
  23. """
  24. The HTTP authorization credentials in the result of using `HTTPBearer` or
  25. `HTTPDigest` in a dependency.
  26. The HTTP authorization header value is split by the first space.
  27. The first part is the `scheme`, the second part is the `credentials`.
  28. For example, in an HTTP Bearer token scheme, the client will send a header
  29. like:
  30. ```
  31. Authorization: Bearer deadbeef12346
  32. ```
  33. In this case:
  34. * `scheme` will have the value `"Bearer"`
  35. * `credentials` will have the value `"deadbeef12346"`
  36. """
  37. scheme: Annotated[
  38. str,
  39. Doc(
  40. """
  41. The HTTP authorization scheme extracted from the header value.
  42. """
  43. ),
  44. ]
  45. credentials: Annotated[
  46. str,
  47. Doc(
  48. """
  49. The HTTP authorization credentials extracted from the header value.
  50. """
  51. ),
  52. ]
  53. class HTTPBase(SecurityBase):
  54. model: HTTPBaseModel
  55. def __init__(
  56. self,
  57. *,
  58. scheme: str,
  59. scheme_name: str | None = None,
  60. description: str | None = None,
  61. auto_error: bool = True,
  62. ):
  63. self.model = HTTPBaseModel(scheme=scheme, description=description)
  64. self.scheme_name = scheme_name or self.__class__.__name__
  65. self.auto_error = auto_error
  66. def make_authenticate_headers(self) -> dict[str, str]:
  67. return {"WWW-Authenticate": f"{self.model.scheme.title()}"}
  68. def make_not_authenticated_error(self) -> HTTPException:
  69. return HTTPException(
  70. status_code=HTTP_401_UNAUTHORIZED,
  71. detail="Not authenticated",
  72. headers=self.make_authenticate_headers(),
  73. )
  74. async def __call__(self, request: Request) -> HTTPAuthorizationCredentials | None:
  75. authorization = request.headers.get("Authorization")
  76. scheme, credentials = get_authorization_scheme_param(authorization)
  77. if not (authorization and scheme and credentials):
  78. if self.auto_error:
  79. raise self.make_not_authenticated_error()
  80. else:
  81. return None
  82. return HTTPAuthorizationCredentials(scheme=scheme, credentials=credentials)
  83. class HTTPBasic(HTTPBase):
  84. """
  85. HTTP Basic authentication.
  86. Ref: https://datatracker.ietf.org/doc/html/rfc7617
  87. ## Usage
  88. Create an instance object and use that object as the dependency in `Depends()`.
  89. The dependency result will be an `HTTPBasicCredentials` object containing the
  90. `username` and the `password`.
  91. Read more about it in the
  92. [FastAPI docs for HTTP Basic Auth](https://fastapi.tiangolo.com/advanced/security/http-basic-auth/).
  93. ## Example
  94. ```python
  95. from typing import Annotated
  96. from fastapi import Depends, FastAPI
  97. from fastapi.security import HTTPBasic, HTTPBasicCredentials
  98. app = FastAPI()
  99. security = HTTPBasic()
  100. @app.get("/users/me")
  101. def read_current_user(credentials: Annotated[HTTPBasicCredentials, Depends(security)]):
  102. return {"username": credentials.username, "password": credentials.password}
  103. ```
  104. """
  105. def __init__(
  106. self,
  107. *,
  108. scheme_name: Annotated[
  109. str | None,
  110. Doc(
  111. """
  112. Security scheme name.
  113. It will be included in the generated OpenAPI (e.g. visible at `/docs`).
  114. """
  115. ),
  116. ] = None,
  117. realm: Annotated[
  118. str | None,
  119. Doc(
  120. """
  121. HTTP Basic authentication realm.
  122. """
  123. ),
  124. ] = None,
  125. description: Annotated[
  126. str | None,
  127. Doc(
  128. """
  129. Security scheme description.
  130. It will be included in the generated OpenAPI (e.g. visible at `/docs`).
  131. """
  132. ),
  133. ] = None,
  134. auto_error: Annotated[
  135. bool,
  136. Doc(
  137. """
  138. By default, if the HTTP Basic authentication is not provided (a
  139. header), `HTTPBasic` will automatically cancel the request and send the
  140. client an error.
  141. If `auto_error` is set to `False`, when the HTTP Basic authentication
  142. is not available, instead of erroring out, the dependency result will
  143. be `None`.
  144. This is useful when you want to have optional authentication.
  145. It is also useful when you want to have authentication that can be
  146. provided in one of multiple optional ways (for example, in HTTP Basic
  147. authentication or in an HTTP Bearer token).
  148. """
  149. ),
  150. ] = True,
  151. ):
  152. self.model = HTTPBaseModel(scheme="basic", description=description)
  153. self.scheme_name = scheme_name or self.__class__.__name__
  154. self.realm = realm
  155. self.auto_error = auto_error
  156. def make_authenticate_headers(self) -> dict[str, str]:
  157. if self.realm:
  158. return {"WWW-Authenticate": f'Basic realm="{self.realm}"'}
  159. return {"WWW-Authenticate": "Basic"}
  160. async def __call__( # type: ignore
  161. self, request: Request
  162. ) -> HTTPBasicCredentials | None:
  163. authorization = request.headers.get("Authorization")
  164. scheme, param = get_authorization_scheme_param(authorization)
  165. if not authorization or scheme.lower() != "basic":
  166. if self.auto_error:
  167. raise self.make_not_authenticated_error()
  168. else:
  169. return None
  170. try:
  171. data = b64decode(param).decode("ascii")
  172. except (ValueError, UnicodeDecodeError, binascii.Error) as e:
  173. raise self.make_not_authenticated_error() from e
  174. username, separator, password = data.partition(":")
  175. if not separator:
  176. raise self.make_not_authenticated_error()
  177. return HTTPBasicCredentials(username=username, password=password)
  178. class HTTPBearer(HTTPBase):
  179. """
  180. HTTP Bearer token authentication.
  181. ## Usage
  182. Create an instance object and use that object as the dependency in `Depends()`.
  183. The dependency result will be an `HTTPAuthorizationCredentials` object containing
  184. the `scheme` and the `credentials`.
  185. ## Example
  186. ```python
  187. from typing import Annotated
  188. from fastapi import Depends, FastAPI
  189. from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
  190. app = FastAPI()
  191. security = HTTPBearer()
  192. @app.get("/users/me")
  193. def read_current_user(
  194. credentials: Annotated[HTTPAuthorizationCredentials, Depends(security)]
  195. ):
  196. return {"scheme": credentials.scheme, "credentials": credentials.credentials}
  197. ```
  198. """
  199. def __init__(
  200. self,
  201. *,
  202. bearerFormat: Annotated[str | None, Doc("Bearer token format.")] = None,
  203. scheme_name: Annotated[
  204. str | None,
  205. Doc(
  206. """
  207. Security scheme name.
  208. It will be included in the generated OpenAPI (e.g. visible at `/docs`).
  209. """
  210. ),
  211. ] = None,
  212. description: Annotated[
  213. str | None,
  214. Doc(
  215. """
  216. Security scheme description.
  217. It will be included in the generated OpenAPI (e.g. visible at `/docs`).
  218. """
  219. ),
  220. ] = None,
  221. auto_error: Annotated[
  222. bool,
  223. Doc(
  224. """
  225. By default, if the HTTP Bearer token is not provided (in an
  226. `Authorization` header), `HTTPBearer` will automatically cancel the
  227. request and send the client an error.
  228. If `auto_error` is set to `False`, when the HTTP Bearer token
  229. is not available, instead of erroring out, the dependency result will
  230. be `None`.
  231. This is useful when you want to have optional authentication.
  232. It is also useful when you want to have authentication that can be
  233. provided in one of multiple optional ways (for example, in an HTTP
  234. Bearer token or in a cookie).
  235. """
  236. ),
  237. ] = True,
  238. ):
  239. self.model = HTTPBearerModel(bearerFormat=bearerFormat, description=description)
  240. self.scheme_name = scheme_name or self.__class__.__name__
  241. self.auto_error = auto_error
  242. async def __call__(self, request: Request) -> HTTPAuthorizationCredentials | None:
  243. authorization = request.headers.get("Authorization")
  244. scheme, credentials = get_authorization_scheme_param(authorization)
  245. if not (authorization and scheme and credentials):
  246. if self.auto_error:
  247. raise self.make_not_authenticated_error()
  248. else:
  249. return None
  250. if scheme.lower() != "bearer":
  251. if self.auto_error:
  252. raise self.make_not_authenticated_error()
  253. else:
  254. return None
  255. return HTTPAuthorizationCredentials(scheme=scheme, credentials=credentials)
  256. class HTTPDigest(HTTPBase):
  257. """
  258. HTTP Digest authentication.
  259. **Warning**: this is only a stub to connect the components with OpenAPI in FastAPI,
  260. but it doesn't implement the full Digest scheme, you would need to subclass it
  261. and implement it in your code.
  262. Ref: https://datatracker.ietf.org/doc/html/rfc7616
  263. ## Usage
  264. Create an instance object and use that object as the dependency in `Depends()`.
  265. The dependency result will be an `HTTPAuthorizationCredentials` object containing
  266. the `scheme` and the `credentials`.
  267. ## Example
  268. ```python
  269. from typing import Annotated
  270. from fastapi import Depends, FastAPI
  271. from fastapi.security import HTTPAuthorizationCredentials, HTTPDigest
  272. app = FastAPI()
  273. security = HTTPDigest()
  274. @app.get("/users/me")
  275. def read_current_user(
  276. credentials: Annotated[HTTPAuthorizationCredentials, Depends(security)]
  277. ):
  278. return {"scheme": credentials.scheme, "credentials": credentials.credentials}
  279. ```
  280. """
  281. def __init__(
  282. self,
  283. *,
  284. scheme_name: Annotated[
  285. str | None,
  286. Doc(
  287. """
  288. Security scheme name.
  289. It will be included in the generated OpenAPI (e.g. visible at `/docs`).
  290. """
  291. ),
  292. ] = None,
  293. description: Annotated[
  294. str | None,
  295. Doc(
  296. """
  297. Security scheme description.
  298. It will be included in the generated OpenAPI (e.g. visible at `/docs`).
  299. """
  300. ),
  301. ] = None,
  302. auto_error: Annotated[
  303. bool,
  304. Doc(
  305. """
  306. By default, if the HTTP Digest is not provided, `HTTPDigest` will
  307. automatically cancel the request and send the client an error.
  308. If `auto_error` is set to `False`, when the HTTP Digest is not
  309. available, instead of erroring out, the dependency result will
  310. be `None`.
  311. This is useful when you want to have optional authentication.
  312. It is also useful when you want to have authentication that can be
  313. provided in one of multiple optional ways (for example, in HTTP
  314. Digest or in a cookie).
  315. """
  316. ),
  317. ] = True,
  318. ):
  319. self.model = HTTPBaseModel(scheme="digest", description=description)
  320. self.scheme_name = scheme_name or self.__class__.__name__
  321. self.auto_error = auto_error
  322. async def __call__(self, request: Request) -> HTTPAuthorizationCredentials | None:
  323. authorization = request.headers.get("Authorization")
  324. scheme, credentials = get_authorization_scheme_param(authorization)
  325. if not (authorization and scheme and credentials):
  326. if self.auto_error:
  327. raise self.make_not_authenticated_error()
  328. else:
  329. return None
  330. if scheme.lower() != "digest":
  331. if self.auto_error:
  332. raise self.make_not_authenticated_error()
  333. else:
  334. return None
  335. return HTTPAuthorizationCredentials(scheme=scheme, credentials=credentials)