mirror of
https://github.com/kuhyx/screen-locker.git
synced 2026-11-10 22:47:13 +01:00
Krzysztof kuhy Rudnicki
bdc405c8c9
Six new modules, all composed or re-exported so no caller changes:
_shutdown.py (384) -> _shutdown_sick_state.py SickDayStateMixin
_wake_alarm.py WakeAlarmMixin
_phone_verification.py (380)-> _adb_transport.py AdbTransportMixin
_http_workout_fetch.py HttpWorkoutFetchMixin
_status_data.py (370) -> _status_types.py the 7 dataclasses
_status_projection.py shutdown projection
+ format_summary_line
Moving a name means moving where it is *read*, so 30-odd mock patch targets
move with it. Most importantly conftest.py's autouse _block_real_network
fixture: ruff correctly dropped the now-unused `socket` import from
_phone_verification, which silently removed the guard's target and would have
let subnet probes reach the real network. It now blocks socket in both modules
that actually own the probes, and both targets are asserted to exist.
976 tests pass at 100% branch coverage; pylint 8.70 -> 8.71; `screen_lock
--status` renders the real week unchanged. File-length gate: 41 -> 35.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01EveB813BQ8abAP68VSqZ5Y
95 lines
3.4 KiB
Python
95 lines
3.4 KiB
Python
"""HTTP fallback for phone verification: find the app's server, fetch JSON.
|
|
|
|
Split out of :mod:`screen_locker._phone_verification` to keep every file under
|
|
the 250-line cap. Composed back into ``PhoneVerificationMixin`` there, so
|
|
callers see no change.
|
|
|
|
Used when adb cannot reach the phone but it is on the same wifi: the workout
|
|
app serves the same JSON over ``WORKOUT_HTTP_PORT``.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from concurrent.futures import ( # pylint: disable=no-name-in-module
|
|
ThreadPoolExecutor,
|
|
as_completed,
|
|
)
|
|
import contextlib
|
|
from http import client as _http_client
|
|
import json
|
|
import logging
|
|
import socket
|
|
|
|
from screen_locker._constants import WORKOUT_HTTP_PORT
|
|
|
|
_HTTPConnection = _http_client.HTTPConnection
|
|
_HTTPException = _http_client.HTTPException
|
|
_HTTP_OK = _http_client.OK
|
|
|
|
_logger = logging.getLogger(__name__)
|
|
|
|
|
|
class HttpWorkoutFetchMixin:
|
|
"""Scans the local subnet for the workout app's HTTP server."""
|
|
|
|
def _scan_for_http_server(self) -> str | None:
|
|
"""Scan local /24 subnet for the workout app HTTP server on port 8765.
|
|
|
|
Returns the first reachable URL or None.
|
|
"""
|
|
prefix = self._get_local_subnet_prefix()
|
|
if prefix is None:
|
|
return None
|
|
|
|
def probe(i: int) -> str | None:
|
|
ip = f"{prefix}.{i}"
|
|
with (
|
|
contextlib.suppress(OSError),
|
|
socket.create_connection((ip, WORKOUT_HTTP_PORT), timeout=0.3),
|
|
):
|
|
return f"http://{ip}:{WORKOUT_HTTP_PORT}/workout"
|
|
return None
|
|
|
|
_logger.info(
|
|
"Scanning %s.1-254:%d for workout app...", prefix, WORKOUT_HTTP_PORT
|
|
)
|
|
with ThreadPoolExecutor(max_workers=64) as executor:
|
|
for future in as_completed(
|
|
executor.submit(probe, i) for i in range(1, 255)
|
|
):
|
|
result = future.result()
|
|
if result is not None:
|
|
return result
|
|
return None
|
|
|
|
def _fetch_http_workout(self) -> dict | None:
|
|
"""Fetch workout JSON from the app's HTTP server on the local network.
|
|
|
|
Uses http.client directly to avoid urllib URL-open security lint rules.
|
|
The URL is always http://<local-ip>:8765/workout — no user input involved.
|
|
"""
|
|
url = self._scan_for_http_server()
|
|
if url is None:
|
|
return None
|
|
# url is always "http://<ip>:<port>/workout" — constructed internally.
|
|
try:
|
|
_, _, hostport = url.partition("://")
|
|
host, _, path = hostport.partition("/")
|
|
hostname, _, port_str = host.partition(":")
|
|
conn = _HTTPConnection(hostname, int(port_str), timeout=5)
|
|
conn.request("GET", f"/{path}")
|
|
resp = conn.getresponse()
|
|
if resp.status != _HTTP_OK:
|
|
return None
|
|
return json.loads(resp.read().decode())
|
|
except (_HTTPException, OSError, ValueError, json.JSONDecodeError) as exc:
|
|
_logger.warning(
|
|
"HTTP fetch of today's workout from %s failed (%s) — no data "
|
|
"from the phone's HTTP server, so verification falls through",
|
|
url,
|
|
exc,
|
|
)
|
|
return None
|
|
|
|
# ── Main verification entry point ─────────────────────────────────────────
|