-
Notifications
You must be signed in to change notification settings - Fork 1
Expand file tree
/
Copy pathinstance.py
More file actions
104 lines (78 loc) · 3.53 KB
/
Copy pathinstance.py
File metadata and controls
104 lines (78 loc) · 3.53 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
"""Query the webapp for trusted details about the current instance.
The container holds a per-instance key at ``/etc/key`` (root-readable only).
Requests signed with this key are verified by the webapp, which resolves all
response fields from its own database. The returned info is therefore
authoritative and cannot be spoofed from inside the container.
"""
import typing as ty
from dataclasses import dataclass
from pathlib import Path
import requests
from itsdangerous import TimedSerializer
from .error import RefUtilsError
KEY_PATH = Path("/etc/key")
INSTANCE_ID_PATH = Path("/etc/instance_id")
INSTANCE_INFO_URL = "http://ssh-reverse-proxy:8000/api/instance/info"
class InstanceInfoError(RefUtilsError):
"""Raised when instance info cannot be retrieved."""
@dataclass
class InstanceInfo:
instance_id: int
is_submission: bool
user_full_name: str
user_mat_num: ty.Optional[str]
is_admin: bool
is_grading_assistant: bool
exercise_short_name: str
exercise_version: int
_cached_info: ty.Optional[InstanceInfo] = None
def _sign_request(instance_id: int, key: bytes, payload: ty.Dict[str, ty.Any]) -> str:
signer = TimedSerializer(key, salt="from-container-to-web")
payload = dict(payload)
payload["instance_id"] = instance_id
return signer.dumps(payload) # type: ignore[no-any-return]
def _fetch_instance_info(timeout_s: float) -> InstanceInfo:
try:
key = KEY_PATH.read_bytes()
instance_id = int(INSTANCE_ID_PATH.read_text().strip())
except OSError as e:
raise InstanceInfoError(f"Failed to read instance credentials: {e}") from e
signed = _sign_request(instance_id, key, {})
try:
resp = requests.post(INSTANCE_INFO_URL, json=signed, timeout=timeout_s)
except requests.RequestException as e:
raise InstanceInfoError(f"Failed to reach webapp: {e}") from e
try:
data = resp.json()
except ValueError as e:
raise InstanceInfoError(f"Invalid JSON response (status={resp.status_code}): {e}") from e
if resp.status_code != 200:
detail = data.get("error", data) if isinstance(data, dict) else data
raise InstanceInfoError(f"Info request failed (status={resp.status_code}): {detail}")
if not isinstance(data, dict):
raise InstanceInfoError(f"Unexpected response body: {data!r}")
try:
return InstanceInfo(**data)
except TypeError as e:
raise InstanceInfoError(f"Response missing or extra fields: {e}") from e
def get_instance_info(timeout_s: float = 10.0, *, refresh: bool = False) -> InstanceInfo:
"""Fetch trusted details about the current instance from the webapp.
All fields originate from the webserver's database; the request is signed
with the instance-specific key at ``/etc/key``, which is not accessible to
the student user. Use this to branch test behavior on user role without
trusting anything the student can tamper with.
The result is cached process-wide on first success (instance metadata does
not change during a test run, and the endpoint is rate-limited). Pass
``refresh=True`` to bypass the cache and force a new fetch. Failed
requests are not cached, so a transient outage does not poison subsequent
calls.
Raises:
InstanceInfoError: if credentials are missing, the webapp is
unreachable, or the response cannot be parsed.
"""
global _cached_info
if _cached_info is not None and not refresh:
return _cached_info
info = _fetch_instance_info(timeout_s)
_cached_info = info
return info