ExternalVendorCode/tahoe-lafs
1
0
Fork 0
mirror of https://github.com/tahoe-lafs/tahoe-lafs.git synced 2025-04-26 13:59:59 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

Use more secure comparison to prevent timing-based side-channel attacks.

Browse Source
This commit is contained in:
Itamar Turner-Trauring 2021-12-20 11:17:27 -05:00
parent a529ba7d5e
commit 291b4e1896
1 changed files with 14 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
src/allmydata/storage/http_server.py
View File

@ -28,10 +28,12 @@ from cbor2 import dumps
from .server import StorageServer from .server import StorageServer
from .http_client import swissnum_auth_header from .http_client import swissnum_auth_header
from ..util.hashutil import timing_safe_compare
class Secrets(Enum): class Secrets(Enum):
"""Different kinds of secrets the client may send.""" """Different kinds of secrets the client may send."""
LEASE_RENEW = "lease-renew-secret" LEASE_RENEW = "lease-renew-secret"
LEASE_CANCEL = "lease-cancel-secret" LEASE_CANCEL = "lease-cancel-secret"
UPLOAD = "upload-secret" UPLOAD = "upload-secret"
@ -41,7 +43,9 @@ class ClientSecretsException(Exception):
"""The client did not send the appropriate secrets.""" """The client did not send the appropriate secrets."""
def _extract_secrets(header_values, required_secrets): # type: (List[str], Set[Secrets]) -> Dict[Secrets, bytes] def _extract_secrets(
header_values, required_secrets
): # type: (List[str], Set[Secrets]) -> Dict[Secrets, bytes]
""" """
Given list of values of ``X-Tahoe-Authorization`` headers, and required Given list of values of ``X-Tahoe-Authorization`` headers, and required
secrets, return dictionary mapping secrets to decoded values. secrets, return dictionary mapping secrets to decoded values.
@ -73,15 +77,21 @@ def _authorization_decorator(required_secrets):
Check the ``Authorization`` header, and (TODO: in later revision of code) Check the ``Authorization`` header, and (TODO: in later revision of code)
extract ``X-Tahoe-Authorization`` headers and pass them in. extract ``X-Tahoe-Authorization`` headers and pass them in.
""" """
def decorator(f): def decorator(f):
@wraps(f) @wraps(f)
def route(self, request, *args, **kwargs): def route(self, request, *args, **kwargs):
if request.requestHeaders.getRawHeaders("Authorization", [None])[0] != str( if not timing_safe_compare(
swissnum_auth_header(self._swissnum), "ascii" request.requestHeaders.getRawHeaders("Authorization", [None])[0].encode(
"utf-8"
),
swissnum_auth_header(self._swissnum),
): ):
request.setResponseCode(http.UNAUTHORIZED) request.setResponseCode(http.UNAUTHORIZED)
return b"" return b""
authorization = request.requestHeaders.getRawHeaders("X-Tahoe-Authorization", []) authorization = request.requestHeaders.getRawHeaders(
"X-Tahoe-Authorization", []
)
try: try:
secrets = _extract_secrets(authorization, required_secrets) secrets = _extract_secrets(authorization, required_secrets)
except ClientSecretsException: except ClientSecretsException:
@ -138,7 +148,6 @@ class HTTPServer(object):
# TODO if data is big, maybe want to use a temporary file eventually... # TODO if data is big, maybe want to use a temporary file eventually...
return dumps(data) return dumps(data)
##### Generic APIs ##### ##### Generic APIs #####
@_authorized_route(_app, set(), "/v1/version", methods=["GET"]) @_authorized_route(_app, set(), "/v1/version", methods=["GET"])