personal memory agent
link: add privacy-scan test and blindness grep evidence
+104
+2
scratch/link-blindness-grep.txt
+2
scratch/link-blindness-grep.txt
+102
tests/link/test_privacy_scan.py
+102
tests/link/test_privacy_scan.py
···
1
+
# SPDX-License-Identifier: AGPL-3.0-only
2
+
# Copyright (c) 2026 sol pbc
3
+
4
+
from __future__ import annotations
5
+
6
+
import base64
7
+
from pathlib import Path
8
+
9
+
import pytest
10
+
11
+
from tests.link.client import Client
12
+
from tests.link.live_helpers import (
13
+
CONVEY_PASSWORD,
14
+
RELAY_URL,
15
+
running_convey_server,
16
+
running_link_service,
17
+
runtime_texts,
18
+
skip_unless_live_relay,
19
+
)
20
+
21
+
pytestmark = pytest.mark.integration
22
+
skip_unless_live_relay()
23
+
24
+
FORBIDDEN_TOKENS = [
25
+
"authorization",
26
+
"bearer ",
27
+
"cookie",
28
+
"ca private",
29
+
"BEGIN PRIVATE",
30
+
"client_cert",
31
+
"home_attestation",
32
+
"x-api-key",
33
+
"payload",
34
+
]
35
+
36
+
37
+
@pytest.mark.asyncio
38
+
@pytest.mark.timeout(60)
39
+
async def test_privacy_scan(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
40
+
tmp_journal = tmp_path / "journal"
41
+
tmp_journal.mkdir()
42
+
monkeypatch.setenv("_SOLSTONE_JOURNAL_OVERRIDE", str(tmp_journal))
43
+
44
+
capture = None
45
+
with (
46
+
running_convey_server(tmp_journal) as base_url,
47
+
running_link_service(tmp_journal) as link_capture,
48
+
):
49
+
capture = link_capture
50
+
identity = Client.pair(base_url, device_label="pytest-device")
51
+
enrolled = Client.enroll_device(RELAY_URL, identity)
52
+
53
+
session = await Client.dial(RELAY_URL, enrolled)
54
+
async with session:
55
+
auth = base64.b64encode(f":{CONVEY_PASSWORD}".encode("utf-8")).decode(
56
+
"ascii"
57
+
)
58
+
status, headers, _body = await session.request(
59
+
"GET",
60
+
"/app/link/api/status",
61
+
headers={"authorization": f"Basic {auth}"},
62
+
)
63
+
assert status == 200
64
+
assert headers["content-type"] == "application/json"
65
+
66
+
assert capture is not None
67
+
texts = runtime_texts(tmp_journal, capture)
68
+
dynamic_tokens = [
69
+
identity.home_attestation,
70
+
enrolled.device_token,
71
+
identity.client_cert_pem[:100],
72
+
identity.home_attestation.split(".")[1],
73
+
]
74
+
75
+
for token in FORBIDDEN_TOKENS:
76
+
_assert_token_absent(texts, token, ignore_case=True)
77
+
for token in dynamic_tokens:
78
+
_assert_token_absent(texts, token, ignore_case=False)
79
+
80
+
81
+
def _assert_token_absent(
82
+
texts: dict[str, str],
83
+
token: str,
84
+
*,
85
+
ignore_case: bool,
86
+
) -> None:
87
+
if not token:
88
+
return
89
+
needle = token.lower() if ignore_case else token
90
+
for source, text in texts.items():
91
+
haystack = text.lower() if ignore_case else text
92
+
match_at = haystack.find(needle)
93
+
if match_at < 0:
94
+
continue
95
+
line_start = text.rfind("\n", 0, match_at) + 1
96
+
line_end = text.find("\n", match_at)
97
+
if line_end < 0:
98
+
line_end = len(text)
99
+
context = text[line_start:line_end]
100
+
raise AssertionError(
101
+
f"forbidden token {token!r} found in {source}: {context!r}"
102
+
)