26369441708

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

Build # 26369441708

Build Type

Pull #1343

github

Committed by

web-flow

Commit Message

Merge 86387626a 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 = {	20✔
47	str(i): stack.enter_context(open(path, "rb")) for i, path in enumerate(paths)
48	}
49	response = requests.post(	20✔
50	url=url,
51	files=upload,
52	data={"id": unique_id, "version": version, "payload": payload},
53	timeout=UPLOAD_TIMEOUT_SECONDS,
54	)
55	response.raise_for_status()	20✔
56	print("[INFO] Upload successful")	20✔
57	except requests.HTTPError as e:	20✔
58	print("[ERROR] Upload failed")	20✔
59	status_code = e.response.status_code if e.response is not None else None	20✔
60	if status_code == 400:	20✔
61	print(	×
62	"[ERROR] HTTP Response Status 400: Client-side error, likely due to improper syntax. "
63	"Please report this to your instructor (and attach the code that caused the error)."
64	)
65	elif status_code == 403:	20✔
66	print(	20✔
67	"[ERROR] HTTP Response Status 403: Authorization is currently required for submission."
68	)
NEW 69	elif status_code == 500:	×
70	print(	×
71	"[ERROR] HTTP Response Status 500: The server ran into a situation it doesn't know how to handle. "
72	)
73	print(	×
74	"Please report this to your instructor (and attach the code that caused the error)."
75	)
NEW 76	elif status_code == 503:	×
77	print(	×
78	"[ERROR] HTTP Response Status 503: The server is not ready to handle your request. "
79	)
80	print("It may be down for maintenance.")	×
81	else:
82	print('[ERROR] Error message: "{}"'.format(e))	×
83	except requests.Timeout:	20✔
84	print("[ERROR] Upload failed")	20✔
85	print("[ERROR] Error message: Connection timed out. The server may be temporarily down.")	20✔
86	except requests.ConnectionError:	20✔
87	print("[ERROR] Upload failed")	20✔
88	print(	20✔
89	"[ERROR] Error message: Could not connect. This may be caused by your firewall, or the server may be "
90	"temporarily down."
91	)
92	except requests.RequestException as e:	20✔
NEW 93	print("[ERROR] Upload failed")	×
NEW 94	print('[ERROR] Error message: "{}"'.format(e))	×
95	except OSError as e:	20✔
96	print("[ERROR] Upload failed")	20✔
97	print(f'[ERROR] Could not read a file selected for upload: "{e}"')	20✔
98
99
100	def get_anonymous_id() -> str:	20✔
101	"""Return an anonymous ID for opt-in data uploads.
102
103	This is a hash of a random local ID so multiple opt-in uploads can be
104	grouped without deriving an identifier from hardware information.
105	"""
106	local_anonymous_id = _get_or_create_local_anonymous_id()	20✔
107	return hashlib.sha512(local_anonymous_id.encode("utf-8")).hexdigest()	20✔
108
109
110	def get_hashed_id() -> str:	20✔
111	"""Return the anonymous upload ID.
112
113	This function is kept as a backwards-compatible alias for older code that
114	imported it directly.
115	"""
116	return get_anonymous_id()	20✔
117
118
119	def _get_or_create_local_anonymous_id() -> str:	20✔
120	"""Return the random local ID used as input for the anonymous upload ID."""
121	global _cached_local_anonymous_id
122
123	anonymous_id_path = _get_anonymous_id_path()	20✔
124	anonymous_id_path_key = str(anonymous_id_path)	20✔
125	if (	20✔
126	_cached_local_anonymous_id is not None
127	and _cached_local_anonymous_id[0] == anonymous_id_path_key
128	):
129	return _cached_local_anonymous_id[1]	20✔
130
131	try:	20✔
132	anonymous_id = anonymous_id_path.read_text(encoding="utf-8").strip()	20✔
133	uuid.UUID(anonymous_id)	20✔
134	return anonymous_id	20✔
135	except (OSError, ValueError):	20✔
136	anonymous_id = str(uuid.uuid4())	20✔
137
138	try:	20✔
139	anonymous_id_path.parent.mkdir(parents=True, exist_ok=True)	20✔
140	anonymous_id_path.write_text(anonymous_id + "\n", encoding="utf-8")	20✔
141	except OSError:	20✔
142	_cached_local_anonymous_id = (anonymous_id_path_key, anonymous_id)	20✔
143	return anonymous_id	20✔
144
145
146	def _iter_error_messages(errors: Iterable[Any]) -> Iterable[Any]:	20✔
147	"""Yield individual messages from current and legacy reporter upload data."""
148	for error_group in errors:	20✔
149	if isinstance(error_group, list):	20✔
150	yield from error_group	20✔
151	elif hasattr(error_group, "code") and hasattr(error_group, "style"):	20✔
152	for error_type in ("code", "style"):	20✔
153	current_type = getattr(error_group, error_type)	20✔
154	for info_set in current_type.values():	20✔
155	yield from info_set.messages	20✔
156	else:
NEW 157	yield error_group	×
158
159
160	def _get_anonymous_id_path() -> Path:	20✔
161	"""Return the local path used to store the anonymous upload ID."""
162	if ANONYMOUS_ID_ENV_VAR in os.environ:	20✔
163	return Path(os.environ[ANONYMOUS_ID_ENV_VAR]).expanduser()	20✔
164
NEW 165	if sys.platform == "win32" and os.environ.get("APPDATA"):	×
NEW 166	return Path(os.environ["APPDATA"]) / "PythonTA" / "anonymous_id"	×
NEW 167	return Path.home() / ".python_ta" / "anonymous_id"	×

pyta-uoft / pyta / 26369441708

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