26369738007

Committed 24 May 2026 06:47PM UTC coverage: 90.393% (-0.5%) from 90.843%

Build # 26369738007

Build Type

Pull #1343

github

Committed by

web-flow

Commit Message

Merge b27bca49d into cbf34b3c5

Pull Request Pull Request #1343: Harden opt-in upload privacy and reliability

Coverage Stats

63 of 72 new or added lines in 1 file covered. (87.5%)

4 existing lines in 1 file now uncovered.

3632 of 4018 relevant lines covered (90.39%)

17.56 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

84.85

/packages/python-ta/src/python_ta/upload.py

from __future__ import annotations

import hashlib
import json
import os
import sys
import uuid
from contextlib import ExitStack
from pathlib import Path
from typing import Any, Iterable

import requests

UPLOAD_TIMEOUT_SECONDS = 5
ANONYMOUS_ID_ENV_VAR = "PYTA_ANONYMOUS_ID_FILE"
_cached_local_anonymous_id: tuple[str, str] | None = None


def errors_to_dict(errors: Iterable[Any]) -> dict[str, list[dict[str, Any]]]:
    """Convert PyTA errors from MessageSet format to a json format Dictionary."""
    error_info = ["msg_id", "msg", "symbol", "module", "category", "line"]
    err_as_dict = {}
    for msg in _iter_error_messages(errors):
        msg_id = getattr(msg, "msg_id", None)
        if msg_id is None:
            continue
        err_as_dict.setdefault(msg_id, []).append(
            {field: getattr(msg, field, None) for field in error_info}
        )
    return err_as_dict


def upload_to_server(
    errors: Iterable[Any], paths: list[str], config: dict[str, Any], url: str, version: str
) -> None:
    """Send POST request to server with formatted data."""
    unique_id = get_anonymous_id()
    errors_dict = errors_to_dict(errors)
    to_json = {"errors": errors_dict}
    if config:  # 'config' is an empty dictionary if the default was used
        to_json["cfg"] = config
    payload = json.dumps(to_json, default=str)

    try:
        with ExitStack() as stack:
            upload = {str(i): stack.enter_context(open(path, "rb")) for i, path in enumerate(paths)}
            response = requests.post(
                url=url,
                files=upload,
                data={"id": unique_id, "version": version, "payload": payload},
                timeout=UPLOAD_TIMEOUT_SECONDS,
            )
        response.raise_for_status()
        print("[INFO] Upload successful")
    except requests.HTTPError as e:
        print("[ERROR] Upload failed")
        status_code = e.response.status_code if e.response is not None else None
        if status_code == 400:
            print(
                "[ERROR] HTTP Response Status 400: Client-side error, likely due to improper syntax. "
                "Please report this to your instructor (and attach the code that caused the error)."
            )
        elif status_code == 403:
            print(
                "[ERROR] HTTP Response Status 403: Authorization is currently required for submission."
            )
        elif status_code == 500:
            print(
                "[ERROR] HTTP Response Status 500: The server ran into a situation it doesn't know how to handle. "
            )
            print(
                "Please report this to your instructor (and attach the code that caused the error)."
            )
        elif status_code == 503:
            print(
                "[ERROR] HTTP Response Status 503: The server is not ready to handle your request. "
            )
            print("It may be down for maintenance.")
        else:
            print('[ERROR] Error message: "{}"'.format(e))
    except requests.Timeout:
        print("[ERROR] Upload failed")
        print("[ERROR] Error message: Connection timed out. The server may be temporarily down.")
    except requests.ConnectionError:
        print("[ERROR] Upload failed")
        print(
            "[ERROR] Error message: Could not connect. This may be caused by your firewall, or the server may be "
            "temporarily down."
        )
    except requests.RequestException as e:
        print("[ERROR] Upload failed")
        print('[ERROR] Error message: "{}"'.format(e))
    except OSError as e:
        print("[ERROR] Upload failed")
        print(f'[ERROR] Could not read a file selected for upload: "{e}"')


def get_anonymous_id() -> str:
    """Return an anonymous ID for opt-in data uploads.

    This is a hash of a random local ID so multiple opt-in uploads can be
    grouped without deriving an identifier from hardware information.
    """
    local_anonymous_id = _get_or_create_local_anonymous_id()
    return hashlib.sha512(local_anonymous_id.encode("utf-8")).hexdigest()


def get_hashed_id() -> str:
    """Return the anonymous upload ID.

    This function is kept as a backwards-compatible alias for older code that
    imported it directly.
    """
    return get_anonymous_id()


def _get_or_create_local_anonymous_id() -> str:
    """Return the random local ID used as input for the anonymous upload ID."""
    global _cached_local_anonymous_id

    anonymous_id_path = _get_anonymous_id_path()
    anonymous_id_path_key = str(anonymous_id_path)
    if (
        _cached_local_anonymous_id is not None
        and _cached_local_anonymous_id[0] == anonymous_id_path_key
    ):
        return _cached_local_anonymous_id[1]

    try:
        anonymous_id = anonymous_id_path.read_text(encoding="utf-8").strip()
        uuid.UUID(anonymous_id)
        return anonymous_id
    except (OSError, ValueError):
        anonymous_id = str(uuid.uuid4())

    try:
        anonymous_id_path.parent.mkdir(parents=True, exist_ok=True)
        anonymous_id_path.write_text(anonymous_id + "\n", encoding="utf-8")
    except OSError:
        _cached_local_anonymous_id = (anonymous_id_path_key, anonymous_id)
    return anonymous_id


def _iter_error_messages(errors: Iterable[Any]) -> Iterable[Any]:
    """Yield individual messages from current and legacy reporter upload data."""
    for error_group in errors:
        if isinstance(error_group, list):
            yield from error_group
        elif hasattr(error_group, "code") and hasattr(error_group, "style"):
            for error_type in ("code", "style"):
                current_type = getattr(error_group, error_type)
                for info_set in current_type.values():
                    yield from info_set.messages
        else:
            yield error_group


def _get_anonymous_id_path() -> Path:
    """Return the local path used to store the anonymous upload ID."""
    if ANONYMOUS_ID_ENV_VAR in os.environ:
        return Path(os.environ[ANONYMOUS_ID_ENV_VAR]).expanduser()

    if sys.platform == "win32" and os.environ.get("APPDATA"):
        return Path(os.environ["APPDATA"]) / "PythonTA" / "anonymous_id"
    return Path.home() / ".python_ta" / "anonymous_id"

1	from __future__ import annotations	20✔
2
3	import hashlib	20✔
4	import json	20✔
5	import os	20✔
6	import sys	20✔
7	import uuid	20✔
8	from contextlib import ExitStack	20✔
9	from pathlib import Path	20✔
10	from typing import Any, Iterable	20✔
11
12	import requests	20✔
13
14	UPLOAD_TIMEOUT_SECONDS = 5	20✔
15	ANONYMOUS_ID_ENV_VAR = "PYTA_ANONYMOUS_ID_FILE"	20✔
16	_cached_local_anonymous_id: tuple[str, str] \| None = None	20✔
17
18
19	def errors_to_dict(errors: Iterable[Any]) -> dict[str, list[dict[str, Any]]]:	20✔
20	"""Convert PyTA errors from MessageSet format to a json format Dictionary."""
21	error_info = ["msg_id", "msg", "symbol", "module", "category", "line"]	20✔
22	err_as_dict = {}	20✔
23	for msg in _iter_error_messages(errors):	20✔
24	msg_id = getattr(msg, "msg_id", None)	20✔
25	if msg_id is None:	20✔
NEW 26	continue	×
27	err_as_dict.setdefault(msg_id, []).append(	20✔
28	{field: getattr(msg, field, None) for field in error_info}
29	)
30	return err_as_dict	20✔
31
32
33	def upload_to_server(	20✔
34	errors: Iterable[Any], paths: list[str], config: dict[str, Any], url: str, version: str
35	) -> None:
36	"""Send POST request to server with formatted data."""
37	unique_id = get_anonymous_id()	20✔
38	errors_dict = errors_to_dict(errors)	20✔
39	to_json = {"errors": errors_dict}	20✔
40	if config: # 'config' is an empty dictionary if the default was used	20✔
41	to_json["cfg"] = config	20✔
42	payload = json.dumps(to_json, default=str)	20✔
43
44	try:	20✔
45	with ExitStack() as stack:	20✔
46	upload = {str(i): stack.enter_context(open(path, "rb")) for i, path in enumerate(paths)}	20✔
47	response = requests.post(	20✔
48	url=url,
49	files=upload,
50	data={"id": unique_id, "version": version, "payload": payload},
51	timeout=UPLOAD_TIMEOUT_SECONDS,
52	)
53	response.raise_for_status()	20✔
54	print("[INFO] Upload successful")	20✔
55	except requests.HTTPError as e:	20✔
56	print("[ERROR] Upload failed")	20✔
57	status_code = e.response.status_code if e.response is not None else None	20✔
58	if status_code == 400:	20✔
59	print(	×
60	"[ERROR] HTTP Response Status 400: Client-side error, likely due to improper syntax. "
61	"Please report this to your instructor (and attach the code that caused the error)."
62	)
63	elif status_code == 403:	20✔
64	print(	20✔
65	"[ERROR] HTTP Response Status 403: Authorization is currently required for submission."
66	)
NEW 67	elif status_code == 500:	×
68	print(	×
69	"[ERROR] HTTP Response Status 500: The server ran into a situation it doesn't know how to handle. "
70	)
71	print(	×
72	"Please report this to your instructor (and attach the code that caused the error)."
73	)
NEW 74	elif status_code == 503:	×
75	print(	×
76	"[ERROR] HTTP Response Status 503: The server is not ready to handle your request. "
77	)
78	print("It may be down for maintenance.")	×
79	else:
80	print('[ERROR] Error message: "{}"'.format(e))	×
81	except requests.Timeout:	20✔
82	print("[ERROR] Upload failed")	20✔
83	print("[ERROR] Error message: Connection timed out. The server may be temporarily down.")	20✔
84	except requests.ConnectionError:	20✔
85	print("[ERROR] Upload failed")	20✔
86	print(	20✔
87	"[ERROR] Error message: Could not connect. This may be caused by your firewall, or the server may be "
88	"temporarily down."
89	)
90	except requests.RequestException as e:	20✔
NEW 91	print("[ERROR] Upload failed")	×
NEW 92	print('[ERROR] Error message: "{}"'.format(e))	×
93	except OSError as e:	20✔
94	print("[ERROR] Upload failed")	20✔
95	print(f'[ERROR] Could not read a file selected for upload: "{e}"')	20✔
96
97
98	def get_anonymous_id() -> str:	20✔
99	"""Return an anonymous ID for opt-in data uploads.
100
101	This is a hash of a random local ID so multiple opt-in uploads can be
102	grouped without deriving an identifier from hardware information.
103	"""
104	local_anonymous_id = _get_or_create_local_anonymous_id()	20✔
105	return hashlib.sha512(local_anonymous_id.encode("utf-8")).hexdigest()	20✔
106
107
108	def get_hashed_id() -> str:	20✔
109	"""Return the anonymous upload ID.
110
111	This function is kept as a backwards-compatible alias for older code that
112	imported it directly.
113	"""
114	return get_anonymous_id()	20✔
115
116
117	def _get_or_create_local_anonymous_id() -> str:	20✔
118	"""Return the random local ID used as input for the anonymous upload ID."""
119	global _cached_local_anonymous_id
120
121	anonymous_id_path = _get_anonymous_id_path()	20✔
122	anonymous_id_path_key = str(anonymous_id_path)	20✔
123	if (	20✔
124	_cached_local_anonymous_id is not None
125	and _cached_local_anonymous_id[0] == anonymous_id_path_key
126	):
127	return _cached_local_anonymous_id[1]	20✔
128
129	try:	20✔
130	anonymous_id = anonymous_id_path.read_text(encoding="utf-8").strip()	20✔
131	uuid.UUID(anonymous_id)	20✔
132	return anonymous_id	20✔
133	except (OSError, ValueError):	20✔
134	anonymous_id = str(uuid.uuid4())	20✔
135
136	try:	20✔
137	anonymous_id_path.parent.mkdir(parents=True, exist_ok=True)	20✔
138	anonymous_id_path.write_text(anonymous_id + "\n", encoding="utf-8")	20✔
139	except OSError:	20✔
140	_cached_local_anonymous_id = (anonymous_id_path_key, anonymous_id)	20✔
141	return anonymous_id	20✔
142
143
144	def _iter_error_messages(errors: Iterable[Any]) -> Iterable[Any]:	20✔
145	"""Yield individual messages from current and legacy reporter upload data."""
146	for error_group in errors:	20✔
147	if isinstance(error_group, list):	20✔
148	yield from error_group	20✔
149	elif hasattr(error_group, "code") and hasattr(error_group, "style"):	20✔
150	for error_type in ("code", "style"):	20✔
151	current_type = getattr(error_group, error_type)	20✔
152	for info_set in current_type.values():	20✔
153	yield from info_set.messages	20✔
154	else:
NEW 155	yield error_group	×
156
157
158	def _get_anonymous_id_path() -> Path:	20✔
159	"""Return the local path used to store the anonymous upload ID."""
160	if ANONYMOUS_ID_ENV_VAR in os.environ:	20✔
161	return Path(os.environ[ANONYMOUS_ID_ENV_VAR]).expanduser()	20✔
162
NEW 163	if sys.platform == "win32" and os.environ.get("APPDATA"):	×
NEW 164	return Path(os.environ["APPDATA"]) / "PythonTA" / "anonymous_id"	×
NEW 165	return Path.home() / ".python_ta" / "anonymous_id"	×

pyta-uoft / pyta / 26369738007

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous