Skip to content

Commit ee0452c

Browse files
committed
Extract Bearer and octet-stream constants
Deduplicates the four 'Bearer ' literals in HTTPActionServer and the three 'application/octet-stream' literals in mime.py — resolves SonarCloud S1192 and keeps the values single-sourced.
1 parent 982ac57 commit ee0452c

2 files changed

Lines changed: 9 additions & 7 deletions

File tree

automation_file/local/mime.py

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@
1212
from pathlib import Path
1313

1414
_SNIFF_LEN = 16
15+
_OCTET_STREAM = "application/octet-stream"
1516
_SIGNATURES: tuple[tuple[bytes, str], ...] = (
1617
(b"\x89PNG\r\n\x1a\n", "image/png"),
1718
(b"\xff\xd8\xff", "image/jpeg"),
@@ -25,7 +26,7 @@
2526
(b"7z\xbc\xaf\x27\x1c", "application/x-7z-compressed"),
2627
(b"Rar!\x1a\x07\x00", "application/vnd.rar"),
2728
(b"Rar!\x1a\x07\x01\x00", "application/vnd.rar"),
28-
(b"RIFF", "application/octet-stream"), # overridden below for wav/webp
29+
(b"RIFF", _OCTET_STREAM), # overridden below for wav/webp
2930
(b"\x00\x00\x00 ftyp", "video/mp4"),
3031
(b"OggS", "application/ogg"),
3132
(b"ID3", "audio/mpeg"),
@@ -46,13 +47,13 @@ def detect_mime(path: str | os.PathLike[str]) -> str:
4647
if guessed:
4748
return guessed
4849
sniffed = _sniff(p)
49-
return sniffed or "application/octet-stream"
50+
return sniffed or _OCTET_STREAM
5051

5152

5253
def detect_from_bytes(data: bytes) -> str:
5354
"""MIME type of a byte blob using magic-byte sniffing."""
5455
mime = _match_signatures(data)
55-
return mime or "application/octet-stream"
56+
return mime or _OCTET_STREAM
5657

5758

5859
def _sniff(path: Path) -> str | None:

automation_file/server/http_server.py

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -42,6 +42,7 @@
4242
_MAX_CONTENT_BYTES = 1 * 1024 * 1024
4343
_PROGRESS_POLL_SECONDS = 1.0
4444
_PROGRESS_MAX_FRAMES = 10_000
45+
_BEARER_PREFIX = "Bearer "
4546

4647

4748
class _HTTPActionHandler(BaseHTTPRequestHandler):
@@ -103,9 +104,9 @@ def _read_payload(self) -> list:
103104
secret: str | None = getattr(self.server, "shared_secret", None)
104105
if secret:
105106
header = self.headers.get("Authorization", "")
106-
if not header.startswith("Bearer "):
107+
if not header.startswith(_BEARER_PREFIX):
107108
raise TCPAuthException("missing bearer token")
108-
if not hmac.compare_digest(header[len("Bearer ") :], secret):
109+
if not hmac.compare_digest(header[len(_BEARER_PREFIX) :], secret):
109110
raise TCPAuthException("bad shared secret")
110111

111112
try:
@@ -142,8 +143,8 @@ def _handle_progress_ws(self) -> None:
142143
secret: str | None = getattr(self.server, "shared_secret", None)
143144
if secret:
144145
header = self.headers.get("Authorization", "")
145-
token_ok = header.startswith("Bearer ") and hmac.compare_digest(
146-
header[len("Bearer ") :], secret
146+
token_ok = header.startswith(_BEARER_PREFIX) and hmac.compare_digest(
147+
header[len(_BEARER_PREFIX) :], secret
147148
)
148149
if not token_ok:
149150
self._send_json(HTTPStatus.UNAUTHORIZED, {"error": "bad shared secret"})

0 commit comments

Comments
 (0)