auth.py 5.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164
  1. """
  2. :mod:`websockets.legacy.auth` provides HTTP Basic Authentication according to
  3. :rfc:`7235` and :rfc:`7617`.
  4. """
  5. import functools
  6. import hmac
  7. import http
  8. from typing import Any, Awaitable, Callable, Iterable, Optional, Tuple, Union, cast
  9. from ..datastructures import Headers
  10. from ..exceptions import InvalidHeader
  11. from ..headers import build_www_authenticate_basic, parse_authorization_basic
  12. from .server import HTTPResponse, WebSocketServerProtocol
  13. __all__ = ["BasicAuthWebSocketServerProtocol", "basic_auth_protocol_factory"]
  14. Credentials = Tuple[str, str]
  15. def is_credentials(value: Any) -> bool:
  16. try:
  17. username, password = value
  18. except (TypeError, ValueError):
  19. return False
  20. else:
  21. return isinstance(username, str) and isinstance(password, str)
  22. class BasicAuthWebSocketServerProtocol(WebSocketServerProtocol):
  23. """
  24. WebSocket server protocol that enforces HTTP Basic Auth.
  25. """
  26. def __init__(
  27. self,
  28. *args: Any,
  29. realm: str,
  30. check_credentials: Callable[[str, str], Awaitable[bool]],
  31. **kwargs: Any,
  32. ) -> None:
  33. self.realm = realm
  34. self.check_credentials = check_credentials
  35. super().__init__(*args, **kwargs)
  36. async def process_request(
  37. self, path: str, request_headers: Headers
  38. ) -> Optional[HTTPResponse]:
  39. """
  40. Check HTTP Basic Auth and return a HTTP 401 or 403 response if needed.
  41. """
  42. try:
  43. authorization = request_headers["Authorization"]
  44. except KeyError:
  45. return (
  46. http.HTTPStatus.UNAUTHORIZED,
  47. [("WWW-Authenticate", build_www_authenticate_basic(self.realm))],
  48. b"Missing credentials\n",
  49. )
  50. try:
  51. username, password = parse_authorization_basic(authorization)
  52. except InvalidHeader:
  53. return (
  54. http.HTTPStatus.UNAUTHORIZED,
  55. [("WWW-Authenticate", build_www_authenticate_basic(self.realm))],
  56. b"Unsupported credentials\n",
  57. )
  58. if not await self.check_credentials(username, password):
  59. return (
  60. http.HTTPStatus.UNAUTHORIZED,
  61. [("WWW-Authenticate", build_www_authenticate_basic(self.realm))],
  62. b"Invalid credentials\n",
  63. )
  64. self.username = username
  65. return await super().process_request(path, request_headers)
  66. def basic_auth_protocol_factory(
  67. realm: str,
  68. credentials: Optional[Union[Credentials, Iterable[Credentials]]] = None,
  69. check_credentials: Optional[Callable[[str, str], Awaitable[bool]]] = None,
  70. create_protocol: Optional[Callable[[Any], BasicAuthWebSocketServerProtocol]] = None,
  71. ) -> Callable[[Any], BasicAuthWebSocketServerProtocol]:
  72. """
  73. Protocol factory that enforces HTTP Basic Auth.
  74. ``basic_auth_protocol_factory`` is designed to integrate with
  75. :func:`~websockets.legacy.server.serve` like this::
  76. websockets.serve(
  77. ...,
  78. create_protocol=websockets.basic_auth_protocol_factory(
  79. realm="my dev server",
  80. credentials=("hello", "iloveyou"),
  81. )
  82. )
  83. ``realm`` indicates the scope of protection. It should contain only ASCII
  84. characters because the encoding of non-ASCII characters is undefined.
  85. Refer to section 2.2 of :rfc:`7235` for details.
  86. ``credentials`` defines hard coded authorized credentials. It can be a
  87. ``(username, password)`` pair or a list of such pairs.
  88. ``check_credentials`` defines a coroutine that checks whether credentials
  89. are authorized. This coroutine receives ``username`` and ``password``
  90. arguments and returns a :class:`bool`.
  91. One of ``credentials`` or ``check_credentials`` must be provided but not
  92. both.
  93. By default, ``basic_auth_protocol_factory`` creates a factory for building
  94. :class:`BasicAuthWebSocketServerProtocol` instances. You can override this
  95. with the ``create_protocol`` parameter.
  96. :param realm: scope of protection
  97. :param credentials: hard coded credentials
  98. :param check_credentials: coroutine that verifies credentials
  99. :raises TypeError: if the credentials argument has the wrong type
  100. """
  101. if (credentials is None) == (check_credentials is None):
  102. raise TypeError("provide either credentials or check_credentials")
  103. if credentials is not None:
  104. if is_credentials(credentials):
  105. credentials_list = [cast(Credentials, credentials)]
  106. elif isinstance(credentials, Iterable):
  107. credentials_list = list(credentials)
  108. if not all(is_credentials(item) for item in credentials_list):
  109. raise TypeError(f"invalid credentials argument: {credentials}")
  110. else:
  111. raise TypeError(f"invalid credentials argument: {credentials}")
  112. credentials_dict = dict(credentials_list)
  113. async def check_credentials(username: str, password: str) -> bool:
  114. try:
  115. expected_password = credentials_dict[username]
  116. except KeyError:
  117. return False
  118. return hmac.compare_digest(expected_password, password)
  119. if create_protocol is None:
  120. # Not sure why mypy cannot figure this out.
  121. create_protocol = cast(
  122. Callable[[Any], BasicAuthWebSocketServerProtocol],
  123. BasicAuthWebSocketServerProtocol,
  124. )
  125. return functools.partial(
  126. create_protocol,
  127. realm=realm,
  128. check_credentials=check_credentials,
  129. )