694 lines
30 KiB
Python
694 lines
30 KiB
Python
import json
|
||
import subprocess
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
|
||
from tests.dashboard_bundle import dashboard
|
||
|
||
|
||
ROOT = Path(__file__).resolve().parents[1]
|
||
SESSION_JS = ROOT / "frontend" / "session.js"
|
||
|
||
|
||
def run_session_scenario(scenario: str) -> dict:
|
||
harness = f"""
|
||
const createSessionBoundary = require({json.dumps(str(SESSION_JS))});
|
||
const state = {{ requests: [], removed: [], deletedDatabases: [], deletionCompleted: false, deletedCaches: [], assigned: '', replaced: [], assignedAfterDeletion: false, replacedAfterDeletion: false, expiredAfterDeletion: false, workerMessages: [], confirmations: [], prompts: [], clearErrors: [], activityListeners: {{}}, now: 100000, responseStatus: 200, responsePayload: {{}}, responses: [] }};
|
||
const storage = {{
|
||
values: new Map([['stackchain.private', 'secret'], ['gitea.preference', 'keep']]),
|
||
get length() {{ return this.values.size; }},
|
||
key(index) {{ return Array.from(this.values.keys())[index] || null; }},
|
||
getItem(key) {{ return this.values.get(key) || null; }},
|
||
setItem(key, value) {{ this.values.set(key, String(value)); }},
|
||
removeItem(key) {{ state.removed.push(key); this.values.delete(key); }},
|
||
}};
|
||
const boundary = createSessionBoundary({{
|
||
cookie: () => 'other=x; stackchain_csrf=csrf-proof; theme=dark',
|
||
origin: 'https://forge.example',
|
||
base: '/dashboard/',
|
||
fetchImpl: async (url, options = {{}}) => {{
|
||
state.requests.push({{ url: String(url), method: options.method || 'GET', headers: Object.fromEntries(new Headers(options.headers || {{}})), body: options.body || null }});
|
||
if (state.failFetch) throw new Error('offline');
|
||
const configured = state.responses.length ? state.responses.shift() : {{ status: state.responseStatus, payload: state.responsePayload }};
|
||
return new Response(JSON.stringify(configured.payload), {{ status: configured.status, headers: {{ 'Content-Type': 'application/json' }} }});
|
||
}},
|
||
localStorage: storage,
|
||
sessionStorage: storage,
|
||
indexedDB: {{ deleteDatabase: name => {{
|
||
state.deletedDatabases.push(name);
|
||
const request = {{ onsuccess: null, onerror: null, onblocked: null, error: null }};
|
||
setTimeout(() => {{
|
||
if (state.failDeletion) {{ request.error = new Error('database blocked'); request.onerror?.(); }}
|
||
else {{ state.deletionCompleted = true; request.onsuccess?.(); }}
|
||
}}, 20);
|
||
return request;
|
||
}} }},
|
||
caches: {{ keys: async () => ['stackchain-dashboard-shell-v15', 'gitea-assets'], delete: async key => {{ state.deletedCaches.push(key); }} }},
|
||
serviceWorker: {{ ready: Promise.resolve({{ active: {{ postMessage: (message, ports = []) => {{ state.workerMessages.push(message); ports[0]?.postMessage({{ok:true}}); }} }} }}) }},
|
||
MessageChannel: class {{ constructor() {{
|
||
const first = {{ onmessage: null, postMessage: data => queueMicrotask(() => second.onmessage?.({{data}})) }};
|
||
const second = {{ onmessage: null, postMessage: data => queueMicrotask(() => first.onmessage?.({{data}})) }};
|
||
this.port1 = first; this.port2 = second;
|
||
}} }},
|
||
location: {{
|
||
assign: value => {{ state.assigned = value; state.assignedAfterDeletion = state.deletionCompleted; }},
|
||
replace: value => state.replaced.push(value),
|
||
}},
|
||
onClearError: error => state.clearErrors.push(error.message),
|
||
onExpired: () => {{ state.expiredAfterDeletion = state.deletionCompleted; }},
|
||
addActivityListener: (type, listener) => {{ state.activityListeners[type] = listener; }},
|
||
now: () => state.now,
|
||
setTimer: (_callback, delay) => {{ state.leaseDelay = delay; return 1; }},
|
||
confirmAction: message => {{ state.confirmations.push(message); return true; }},
|
||
promptAuthorization: details => {{ state.prompts.push(details); return 'correct horse battery staple'; }},
|
||
}});
|
||
(async () => {{ {scenario} }})().catch(error => {{ console.error(error); process.exit(1); }});
|
||
"""
|
||
result = subprocess.run(["node", "-e", harness], text=True, capture_output=True, check=True)
|
||
return json.loads(result.stdout)
|
||
|
||
|
||
def test_same_origin_api_fetch_times_out_even_when_fetch_ignores_abort():
|
||
script = f"""
|
||
const createSessionBoundary = require({json.dumps(str(SESSION_JS))});
|
||
let aborted = false;
|
||
const boundary = createSessionBoundary({{
|
||
cookie: () => '',
|
||
origin: 'https://forge.example',
|
||
base: '/dashboard/',
|
||
requestTimeoutMs: 10,
|
||
fetchImpl: (_url, options = {{}}) => new Promise(resolve => {{
|
||
options.signal?.addEventListener('abort', () => {{ aborted = true; }});
|
||
setTimeout(() => resolve(new Response('{{}}', {{status:200}})), 80);
|
||
}}),
|
||
location: {{ replace: () => {{}} }},
|
||
}});
|
||
(async () => {{
|
||
const started = Date.now();
|
||
try {{ await boundary.fetch('/dashboard/api/v1/context'); }}
|
||
catch (error) {{
|
||
process.stdout.write(JSON.stringify({{
|
||
name:error.name, message:error.message, method:error.method,
|
||
outcome:error.outcome, aborted, elapsed:Date.now()-started,
|
||
}}));
|
||
}}
|
||
}})().catch(error => {{ console.error(error); process.exit(1); }});
|
||
"""
|
||
result = subprocess.run(
|
||
["node", "-e", script], text=True, capture_output=True, check=True, timeout=2
|
||
)
|
||
output = json.loads(result.stdout)
|
||
|
||
assert output["name"] == "TimeoutError"
|
||
assert output["message"] == "Request timed out. Try again."
|
||
assert output["method"] == "GET"
|
||
assert output["outcome"] == "unknown"
|
||
assert output["aborted"] is True
|
||
assert output["elapsed"] < 70
|
||
|
||
|
||
def test_caller_abort_cancels_same_origin_api_fetch_without_waiting_for_deadline():
|
||
script = f"""
|
||
const createSessionBoundary = require({json.dumps(str(SESSION_JS))});
|
||
const caller = new AbortController();
|
||
let requestAborted = false;
|
||
const boundary = createSessionBoundary({{
|
||
cookie: () => '', origin:'https://forge.example', base:'/dashboard/', requestTimeoutMs:1000,
|
||
fetchImpl: (_url, options = {{}}) => new Promise(() => {{
|
||
options.signal.addEventListener('abort', () => {{ requestAborted = true; }});
|
||
}}),
|
||
location: {{ replace: () => {{}} }},
|
||
}});
|
||
(async () => {{
|
||
const started = Date.now();
|
||
const pending = boundary.fetch('/dashboard/api/v1/context', {{signal:caller.signal}});
|
||
setTimeout(() => caller.abort(new DOMException('Superseded', 'AbortError')), 10);
|
||
try {{ await pending; }} catch (error) {{
|
||
process.stdout.write(JSON.stringify({{name:error.name, message:error.message, source:error.source, requestAborted, elapsed:Date.now()-started}}));
|
||
}}
|
||
}})().catch(error => {{ console.error(error); process.exit(1); }});
|
||
"""
|
||
result = subprocess.run(
|
||
["node", "-e", script], text=True, capture_output=True, check=True, timeout=2
|
||
)
|
||
output = json.loads(result.stdout)
|
||
|
||
assert output["name"] == "AbortError"
|
||
assert output["message"] == "Superseded"
|
||
assert output["source"] == "caller"
|
||
assert output["requestAborted"] is True
|
||
assert output["elapsed"] < 100
|
||
|
||
|
||
def test_request_object_abort_signal_is_composed_with_the_api_deadline():
|
||
script = f"""
|
||
const createSessionBoundary = require({json.dumps(str(SESSION_JS))});
|
||
const caller = new AbortController();
|
||
const boundary = createSessionBoundary({{
|
||
cookie: () => '', origin:'https://forge.example', base:'/dashboard/', requestTimeoutMs:1000,
|
||
fetchImpl: () => new Promise(() => {{}}), location: {{replace:() => {{}}}},
|
||
}});
|
||
(async () => {{
|
||
const request = new Request('https://forge.example/dashboard/api/v1/context', {{signal:caller.signal}});
|
||
const pending = boundary.fetch(request);
|
||
setTimeout(() => caller.abort(new DOMException('Request superseded', 'AbortError')), 5);
|
||
try {{ await pending; }} catch (error) {{
|
||
process.stdout.write(JSON.stringify({{name:error.name, message:error.message, source:error.source}}));
|
||
}}
|
||
}})().catch(error => {{ console.error(error); process.exit(1); }});
|
||
"""
|
||
result = subprocess.run(
|
||
["node", "-e", script], text=True, capture_output=True, check=True, timeout=2
|
||
)
|
||
|
||
assert json.loads(result.stdout) == {
|
||
"name": "AbortError",
|
||
"message": "Request superseded",
|
||
"source": "caller",
|
||
}
|
||
|
||
|
||
def test_mutation_timeout_warns_that_the_server_outcome_may_be_ambiguous():
|
||
script = f"""
|
||
const createSessionBoundary = require({json.dumps(str(SESSION_JS))});
|
||
const boundary = createSessionBoundary({{
|
||
cookie: () => 'stackchain_csrf=proof', origin:'https://forge.example', base:'/dashboard/', requestTimeoutMs:5,
|
||
fetchImpl: () => new Promise(() => {{}}), location: {{replace:() => {{}}}},
|
||
}});
|
||
(async () => {{
|
||
try {{ await boundary.fetch('/dashboard/api/v1/repos/stackchain/app/issues/9', {{method:'PATCH'}}); }}
|
||
catch (error) {{ process.stdout.write(JSON.stringify({{name:error.name, message:error.message, method:error.method, safeToRetry:error.safeToRetry}})); }}
|
||
}})().catch(error => {{ console.error(error); process.exit(1); }});
|
||
"""
|
||
result = subprocess.run(
|
||
["node", "-e", script], text=True, capture_output=True, check=True, timeout=2
|
||
)
|
||
output = json.loads(result.stdout)
|
||
|
||
assert output == {
|
||
"name": "TimeoutError",
|
||
"message": "Request timed out. Refresh to verify the outcome before retrying.",
|
||
"method": "PATCH",
|
||
"safeToRetry": False,
|
||
}
|
||
|
||
|
||
def test_fresh_authorization_request_uses_the_same_deadline_boundary():
|
||
script = f"""
|
||
const createSessionBoundary = require({json.dumps(str(SESSION_JS))});
|
||
let calls = 0;
|
||
const boundary = createSessionBoundary({{
|
||
cookie: () => 'stackchain_csrf=proof', origin:'https://forge.example', base:'/dashboard/', requestTimeoutMs:5,
|
||
fetchImpl: () => {{
|
||
calls += 1;
|
||
if (calls === 1) return Promise.resolve(new Response(JSON.stringify({{detail:{{code:'step_up_required', action:'merge_pull', target:'stackchain/app#9'}}}}), {{status:428, headers:{{'Content-Type':'application/json'}}}}));
|
||
return new Promise(() => {{}});
|
||
}},
|
||
promptAuthorization: async () => 'operator-token', location: {{replace:() => {{}}}},
|
||
}});
|
||
(async () => {{
|
||
try {{ await boundary.fetch('/dashboard/api/v1/repos/stackchain/app/pulls/9/merge', {{method:'POST'}}); }}
|
||
catch (error) {{ process.stdout.write(JSON.stringify({{name:error.name, phase:error.phase, calls}})); }}
|
||
}})().catch(error => {{ console.error(error); process.exit(1); }});
|
||
"""
|
||
result = subprocess.run(
|
||
["node", "-e", script], text=True, capture_output=True, check=True, timeout=2
|
||
)
|
||
|
||
assert json.loads(result.stdout) == {
|
||
"name": "TimeoutError",
|
||
"phase": "fresh-authorization",
|
||
"calls": 2,
|
||
}
|
||
|
||
|
||
def test_mutating_same_origin_fetch_receives_csrf_proof():
|
||
result = run_session_scenario(
|
||
"""
|
||
await boundary.fetch('/dashboard/api/v1/notifications/7/read', { method: 'PATCH' });
|
||
process.stdout.write(JSON.stringify(state));
|
||
"""
|
||
)
|
||
|
||
assert result["requests"][0]["headers"]["x-csrf-token"] == "csrf-proof"
|
||
|
||
|
||
def test_user_activity_heartbeat_is_throttled_across_pointer_keyboard_and_touch_events():
|
||
result = run_session_scenario(
|
||
"""
|
||
boundary.startActivityHeartbeat();
|
||
await state.activityListeners.pointerdown();
|
||
await state.activityListeners.keydown();
|
||
state.now += 60000;
|
||
await state.activityListeners.touchstart();
|
||
process.stdout.write(JSON.stringify(state));
|
||
"""
|
||
)
|
||
|
||
assert [request["url"] for request in result["requests"]] == [
|
||
"/dashboard/api/v1/session/activity",
|
||
"/dashboard/api/v1/session/activity",
|
||
]
|
||
assert all(request["method"] == "POST" for request in result["requests"])
|
||
assert all(
|
||
request["headers"]["x-csrf-token"] == "csrf-proof"
|
||
for request in result["requests"]
|
||
)
|
||
|
||
|
||
def test_authenticated_session_status_persists_offline_lease_for_worker_and_expiry_timer():
|
||
result = run_session_scenario(
|
||
"""
|
||
state.responsePayload = {authenticated:true, csrf_token:'csrf-proof', expires_at:4102444800};
|
||
const valid = await boundary.refreshOfflineLease();
|
||
state.valid = valid;
|
||
state.savedExpiry = storage.getItem('stackchain.session-expires-at');
|
||
process.stdout.write(JSON.stringify(state));
|
||
"""
|
||
)
|
||
|
||
assert result["valid"] is True
|
||
assert result["savedExpiry"] == "4102444800"
|
||
assert result["workerMessages"] == [
|
||
{"type": "stackchain-session-lease", "expiresAt": 4102444800}
|
||
]
|
||
assert result["leaseDelay"] > 0
|
||
|
||
|
||
def test_expired_stored_lease_purges_private_work_when_session_status_is_offline():
|
||
result = run_session_scenario(
|
||
"""
|
||
storage.setItem('stackchain.session-expires-at', '1');
|
||
state.failFetch = true;
|
||
const valid = await boundary.refreshOfflineLease();
|
||
state.valid = valid;
|
||
state.remaining = Array.from(storage.values.keys());
|
||
state.replacedAfterDeletion = state.deletionCompleted && state.replaced.length === 1;
|
||
process.stdout.write(JSON.stringify(state));
|
||
"""
|
||
)
|
||
|
||
assert result["valid"] is False
|
||
assert result["remaining"] == ["gitea.preference"]
|
||
assert result["deletedDatabases"] == ["stackchain-background-outbox-v1"]
|
||
assert result["deletedCaches"] == ["stackchain-dashboard-shell-v15"]
|
||
assert result["replaced"] == ["/dashboard/login?reason=session-expired"]
|
||
assert result["replacedAfterDeletion"] is True
|
||
|
||
|
||
def test_enroll_passkey_bootstraps_with_fresh_authorization_and_web_authentication():
|
||
script = f"""
|
||
const createSessionBoundary=require({json.dumps(str(SESSION_JS))});
|
||
const requests=[];
|
||
const responses=[
|
||
{{status:428,payload:{{detail:{{code:'step_up_required',action:'enroll_passkey',target:'current_device'}}}}}},
|
||
{{status:201,payload:{{grant:'bootstrap-grant',expires_in:90}}}},
|
||
{{status:201,payload:{{challenge:'AQID',rp:{{id:'forge.example',name:'Stackchain Dashboard'}},user:{{id:'BAUG',name:'stackchain-operator',displayName:'Stackchain operator'}},pubKeyCredParams:[],excludeCredentials:[],authenticatorSelection:{{userVerification:'required'}}}}}},
|
||
{{status:201,payload:{{enrolled:true}}}},
|
||
];
|
||
const credential={{id:'new-passkey',type:'public-key',rawId:Uint8Array.from([1]).buffer,response:{{attestationObject:Uint8Array.from([2]).buffer,clientDataJSON:Uint8Array.from([3]).buffer,getTransports:()=>['internal']}}}};
|
||
const boundary=createSessionBoundary({{
|
||
cookie:()=> 'stackchain_csrf=proof',origin:'https://forge.example',base:'/dashboard/',
|
||
credentials:{{create:async options=>{{globalThis.creation=options.publicKey;return credential;}}}},
|
||
promptAuthorization:async()=> 'recovery-token',
|
||
fetchImpl:async(url,options={{}})=>{{requests.push({{url:String(url),body:options.body||null,grant:new Headers(options.headers||{{}}).get('X-Step-Up-Grant')}});const item=responses.shift();return new Response(JSON.stringify(item.payload),{{status:item.status,headers:{{'Content-Type':'application/json'}}}});}},
|
||
location:{{replace:()=>{{}}}},
|
||
}});
|
||
(async()=>{{const enrolled=await boundary.enrollPasskey();process.stdout.write(JSON.stringify({{enrolled,requests,creation:globalThis.creation}}));}})().catch(error=>{{console.error(error);process.exit(1);}});
|
||
"""
|
||
result = subprocess.run(
|
||
["node", "-e", script], text=True, capture_output=True, check=True
|
||
)
|
||
output = json.loads(result.stdout)
|
||
|
||
assert output["enrolled"] is True
|
||
assert output["creation"]["challenge"] == {"0": 1, "1": 2, "2": 3}
|
||
assert output["creation"]["user"]["id"] == {"0": 4, "1": 5, "2": 6}
|
||
assert output["requests"][2]["grant"] == "bootstrap-grant"
|
||
verification = json.loads(output["requests"][3]["body"])
|
||
assert verification["challenge"] == "AQID"
|
||
assert verification["credential"]["response"]["transports"] == ["internal"]
|
||
|
||
|
||
def test_high_impact_fetch_uses_passkey_before_access_token_fallback():
|
||
script = f"""
|
||
const createSessionBoundary = require({json.dumps(str(SESSION_JS))});
|
||
const requests=[];
|
||
const responses=[
|
||
{{status:428,payload:{{detail:{{code:'step_up_required',action:'close_issue',target:'stackchain/api#7'}}}}}},
|
||
{{status:200,payload:{{challenge:'AQID',rpId:'forge.example',userVerification:'required',allowCredentials:[{{type:'public-key',id:'BAUG'}}]}}}},
|
||
{{status:201,payload:{{grant:'passkey-grant',expires_in:90}}}},
|
||
{{status:200,payload:{{closed:true}}}},
|
||
];
|
||
let prompted=false;
|
||
const credential={{id:'credential',type:'public-key',rawId:Uint8Array.from([4,5,6]).buffer,response:{{authenticatorData:Uint8Array.from([7]).buffer,clientDataJSON:Uint8Array.from([8]).buffer,signature:Uint8Array.from([9]).buffer,userHandle:null}}}};
|
||
const boundary=createSessionBoundary({{
|
||
cookie:()=> 'stackchain_csrf=proof',origin:'https://forge.example',base:'/dashboard/',
|
||
credentials:{{get:async()=>credential}},
|
||
promptAuthorization:async()=>{{prompted=true;return 'must-not-be-requested';}},
|
||
fetchImpl:async(url,options={{}})=>{{requests.push({{url:String(url),body:options.body||null,grant:new Headers(options.headers||{{}}).get('X-Step-Up-Grant')}});const item=responses.shift();return new Response(JSON.stringify(item.payload),{{status:item.status,headers:{{'Content-Type':'application/json'}}}});}},
|
||
location:{{replace:()=>{{}}}},
|
||
}});
|
||
(async()=>{{const response=await boundary.fetch('/dashboard/api/v1/repos/stackchain/api/issues/7/close',{{method:'PATCH'}});process.stdout.write(JSON.stringify({{status:response.status,requests,prompted}}));}})().catch(error=>{{console.error(error);process.exit(1);}});
|
||
"""
|
||
result = subprocess.run(
|
||
["node", "-e", script], text=True, capture_output=True, check=True
|
||
)
|
||
output = json.loads(result.stdout)
|
||
|
||
assert output["status"] == 200
|
||
assert output["prompted"] is False
|
||
assert output["requests"][1]["url"].endswith("/passkeys/authorization/options")
|
||
assert json.loads(output["requests"][1]["body"]) == {
|
||
"action": "close_issue",
|
||
"target": "stackchain/api#7",
|
||
}
|
||
assert output["requests"][2]["url"].endswith("/passkeys/authorization/verify")
|
||
assert output["requests"][3]["grant"] == "passkey-grant"
|
||
|
||
|
||
def test_high_impact_fetch_prompts_once_and_retries_original_request_with_grant():
|
||
result = run_session_scenario(
|
||
"""
|
||
state.responses = [
|
||
{status:428, payload:{detail:{code:'step_up_required', action:'merge_pull', target:'stackchain/api#7'}}},
|
||
{status:201, payload:{grant:'one-time-grant', expires_in:90}},
|
||
{status:200, payload:{merged:true}},
|
||
];
|
||
const original = { method:'POST', headers:{'Content-Type':'application/json'}, body:JSON.stringify({expected_head_sha:'abc123'}) };
|
||
const response = await boundary.fetch('/dashboard/api/v1/repos/stackchain/api/pulls/7/merge', original);
|
||
state.finalStatus = response.status;
|
||
process.stdout.write(JSON.stringify(state));
|
||
"""
|
||
)
|
||
|
||
assert result["finalStatus"] == 200
|
||
assert result["prompts"] == [
|
||
{"action": "merge_pull", "target": "stackchain/api#7"}
|
||
]
|
||
assert [request["url"] for request in result["requests"]] == [
|
||
"/dashboard/api/v1/repos/stackchain/api/pulls/7/merge",
|
||
"/dashboard/api/v1/fresh-authorization",
|
||
"/dashboard/api/v1/repos/stackchain/api/pulls/7/merge",
|
||
]
|
||
assert json.loads(result["requests"][1]["body"]) == {
|
||
"access_token": "correct horse battery staple",
|
||
"action": "merge_pull",
|
||
"target": "stackchain/api#7",
|
||
}
|
||
assert result["requests"][2]["headers"]["x-step-up-grant"] == "one-time-grant"
|
||
assert result["requests"][2]["body"] == result["requests"][0]["body"]
|
||
|
||
|
||
def test_same_origin_unauthorized_response_purges_private_work_before_replacing_dashboard_once():
|
||
result = run_session_scenario(
|
||
"""
|
||
state.responseStatus = 401;
|
||
await Promise.all([
|
||
boundary.fetch('/dashboard/api/v1/live'),
|
||
boundary.fetch('https://forge.example/dashboard/api/v1/context'),
|
||
]);
|
||
state.remaining = Array.from(storage.values.keys());
|
||
state.replacedAfterDeletion = state.deletionCompleted && state.replaced.length === 1;
|
||
process.stdout.write(JSON.stringify(state));
|
||
"""
|
||
)
|
||
|
||
assert result["replaced"] == [
|
||
"/dashboard/login?reason=session-expired"
|
||
]
|
||
assert result["remaining"] == ["gitea.preference"]
|
||
assert result["deletedDatabases"] == ["stackchain-background-outbox-v1"]
|
||
assert result["deletedCaches"] == ["stackchain-dashboard-shell-v15"]
|
||
assert result["workerMessages"] == [{"type": "stackchain-purge-outbox"}]
|
||
assert result["expiredAfterDeletion"] is True
|
||
assert result["replacedAfterDeletion"] is True
|
||
|
||
|
||
def test_remotely_revoked_response_clears_private_work_before_replacing_dashboard():
|
||
result = run_session_scenario(
|
||
"""
|
||
state.responseStatus = 401;
|
||
state.responsePayload = {detail:'Authentication required', code:'session_revoked'};
|
||
await boundary.fetch('/dashboard/api/v1/live');
|
||
state.remaining = Array.from(storage.values.keys());
|
||
state.replacedAfterDeletion = state.deletionCompleted && state.replaced.length === 1;
|
||
process.stdout.write(JSON.stringify(state));
|
||
"""
|
||
)
|
||
|
||
assert result["remaining"] == ["gitea.preference"]
|
||
assert result["deletedDatabases"] == ["stackchain-background-outbox-v1"]
|
||
assert result["deletedCaches"] == ["stackchain-dashboard-shell-v15"]
|
||
assert result["workerMessages"] == [{"type": "stackchain-purge-outbox"}]
|
||
assert result["replaced"] == ["/dashboard/login?reason=session-revoked"]
|
||
assert result["replacedAfterDeletion"] is True
|
||
|
||
|
||
def test_idle_response_locks_without_clearing_private_queued_work():
|
||
result = run_session_scenario(
|
||
"""
|
||
state.responseStatus = 401;
|
||
state.responsePayload = {detail:'Authentication required', code:'session_idle'};
|
||
await boundary.fetch('/dashboard/api/v1/live');
|
||
state.remaining = Array.from(storage.values.keys());
|
||
process.stdout.write(JSON.stringify(state));
|
||
"""
|
||
)
|
||
|
||
assert result["remaining"] == ["stackchain.private", "gitea.preference"]
|
||
assert result["deletedDatabases"] == []
|
||
assert result["deletedCaches"] == []
|
||
assert result["workerMessages"] == []
|
||
assert result["replaced"] == ["/dashboard/login?reason=session-idle"]
|
||
|
||
|
||
def test_worker_revocation_message_clears_window_storage_before_login():
|
||
result = run_session_scenario(
|
||
"""
|
||
await boundary.handleServiceWorkerMessage({data:{type:'stackchain-session-revoked'}});
|
||
state.remaining = Array.from(storage.values.keys());
|
||
process.stdout.write(JSON.stringify(state));
|
||
"""
|
||
)
|
||
|
||
assert result["remaining"] == ["gitea.preference"]
|
||
assert result["replaced"] == ["/dashboard/login?reason=session-revoked"]
|
||
|
||
|
||
def test_worker_idle_message_locks_without_clearing_window_storage():
|
||
result = run_session_scenario(
|
||
"""
|
||
await boundary.handleServiceWorkerMessage({data:{type:'stackchain-session-idle'}});
|
||
state.remaining = Array.from(storage.values.keys());
|
||
process.stdout.write(JSON.stringify(state));
|
||
"""
|
||
)
|
||
|
||
assert result["remaining"] == ["stackchain.private", "gitea.preference"]
|
||
assert result["replaced"] == ["/dashboard/login?reason=session-idle"]
|
||
|
||
|
||
def test_worker_expiry_message_clears_window_storage_before_expired_login():
|
||
result = run_session_scenario(
|
||
"""
|
||
await boundary.handleServiceWorkerMessage({data:{type:'stackchain-session-expired'}});
|
||
state.remaining = Array.from(storage.values.keys());
|
||
state.replacedAfterDeletion = state.deletionCompleted && state.replaced.length === 1;
|
||
process.stdout.write(JSON.stringify(state));
|
||
"""
|
||
)
|
||
|
||
assert result["remaining"] == ["gitea.preference"]
|
||
assert result["replaced"] == ["/dashboard/login?reason=session-expired"]
|
||
assert result["replacedAfterDeletion"] is True
|
||
|
||
|
||
def test_cross_origin_unauthorized_response_does_not_expire_dashboard_session():
|
||
result = run_session_scenario(
|
||
"""
|
||
state.responseStatus = 401;
|
||
await boundary.fetch('https://untrusted.example/api/private');
|
||
process.stdout.write(JSON.stringify(state));
|
||
"""
|
||
)
|
||
|
||
assert result["replaced"] == []
|
||
|
||
|
||
def test_sign_out_clears_only_dashboard_private_device_state():
|
||
result = run_session_scenario(
|
||
"""
|
||
await boundary.signOut();
|
||
state.remaining = Array.from(storage.values.keys());
|
||
process.stdout.write(JSON.stringify(state));
|
||
"""
|
||
)
|
||
|
||
request = result["requests"][0]
|
||
assert request["url"] == "/dashboard/api/v1/session"
|
||
assert request["method"] == "DELETE"
|
||
assert request["headers"]["x-csrf-token"] == "csrf-proof"
|
||
assert "stackchain.private" in result["removed"]
|
||
assert result["remaining"] == ["gitea.preference"]
|
||
assert result["deletedDatabases"] == ["stackchain-background-outbox-v1"]
|
||
assert result["deletedCaches"] == ["stackchain-dashboard-shell-v15"]
|
||
assert result["assigned"] == "/dashboard/login"
|
||
assert result["assignedAfterDeletion"] is True
|
||
|
||
|
||
def test_sign_out_stays_on_page_and_reports_failed_private_outbox_deletion():
|
||
result = run_session_scenario(
|
||
"""
|
||
state.failDeletion = true;
|
||
try { await boundary.signOut(); } catch (error) { state.signOutError = error.message; }
|
||
process.stdout.write(JSON.stringify(state));
|
||
"""
|
||
)
|
||
|
||
assert result["assigned"] == ""
|
||
assert result["clearErrors"] == ["Could not clear private queued work from this device."]
|
||
assert result["signOutError"] == "Could not clear private queued work from this device."
|
||
|
||
|
||
def test_sign_out_all_devices_requires_confirmation_and_uses_global_endpoint():
|
||
result = run_session_scenario(
|
||
"""
|
||
await boundary.signOutAllDevices();
|
||
state.remaining = Array.from(storage.values.keys());
|
||
process.stdout.write(JSON.stringify(state));
|
||
"""
|
||
)
|
||
|
||
assert result["confirmations"] == [
|
||
"Sign out every device? You will need to sign in again everywhere."
|
||
]
|
||
request = result["requests"][0]
|
||
assert request["url"] == "/dashboard/api/v1/sessions"
|
||
assert request["method"] == "DELETE"
|
||
assert request["headers"]["x-csrf-token"] == "csrf-proof"
|
||
assert result["remaining"] == ["gitea.preference"]
|
||
assert result["assigned"] == "/dashboard/login"
|
||
|
||
|
||
def test_active_devices_can_be_loaded_and_one_remote_device_revoked():
|
||
result = run_session_scenario(
|
||
"""
|
||
state.responsePayload = {devices:[
|
||
{management_id:'remote-id-123456789', device_label:'Pixel <script>', created_at:1000, expires_at:2000, current:false},
|
||
{management_id:'current-id-12345678', device_label:'Work laptop', created_at:1100, expires_at:2100, current:true},
|
||
]};
|
||
const devices = await boundary.listActiveDevices();
|
||
await boundary.revokeActiveDevice(devices[0]);
|
||
state.devices = devices;
|
||
process.stdout.write(JSON.stringify(state));
|
||
"""
|
||
)
|
||
|
||
assert result["devices"][0]["device_label"] == "Pixel <script>"
|
||
assert result["requests"][0]["url"] == "/dashboard/api/v1/sessions"
|
||
assert result["requests"][1]["url"] == (
|
||
"/dashboard/api/v1/sessions/remote-id-123456789"
|
||
)
|
||
assert result["requests"][1]["method"] == "DELETE"
|
||
assert result["requests"][1]["headers"]["x-csrf-token"] == "csrf-proof"
|
||
assert result["confirmations"] == ["Sign out Pixel <script>?"]
|
||
|
||
|
||
def test_enrolled_passkeys_can_be_loaded_and_selectively_removed():
|
||
result = run_session_scenario(
|
||
"""
|
||
state.responses = [
|
||
{status:200,payload:{passkeys:[
|
||
{management_id:'passkey-id-12345678', device_label:'Old phone <script>', created_at:1000, active:false, current:false},
|
||
]}},
|
||
{status:200,payload:{revoked:true,current_session:false,session_revoked:false}},
|
||
];
|
||
const passkeys = await boundary.listPasskeys();
|
||
const outcome = await boundary.revokePasskey(passkeys[0]);
|
||
state.passkeys = passkeys;
|
||
state.outcome = outcome;
|
||
process.stdout.write(JSON.stringify(state));
|
||
"""
|
||
)
|
||
|
||
assert result["passkeys"][0]["device_label"] == "Old phone <script>"
|
||
assert result["outcome"] == {
|
||
"revoked": True,
|
||
"current_session": False,
|
||
"session_revoked": False,
|
||
}
|
||
assert result["requests"][0]["url"] == "/dashboard/api/v1/passkeys"
|
||
assert result["requests"][1]["url"] == (
|
||
"/dashboard/api/v1/passkeys/passkey-id-12345678"
|
||
)
|
||
assert result["requests"][1]["method"] == "DELETE"
|
||
assert result["requests"][1]["headers"]["x-csrf-token"] == "csrf-proof"
|
||
assert result["confirmations"] == [
|
||
"Remove the passkey for Old phone <script>? This cannot be undone."
|
||
]
|
||
|
||
|
||
def test_security_activity_pages_are_loaded_with_stable_cursors():
|
||
result = run_session_scenario(
|
||
"""
|
||
state.responsePayload = {events:[{id:7,kind:'sign_in',method:'passkey',device_label:'Phone',target:'dashboard',created_at:1000}],next_cursor:7};
|
||
const first = await boundary.listSecurityEvents();
|
||
const second = await boundary.listSecurityEvents(first.next_cursor);
|
||
state.first = first;
|
||
process.stdout.write(JSON.stringify(state));
|
||
"""
|
||
)
|
||
|
||
assert result["first"]["events"][0]["method"] == "passkey"
|
||
assert [request["url"] for request in result["requests"]] == [
|
||
"/dashboard/api/v1/security-events?limit=25",
|
||
"/dashboard/api/v1/security-events?limit=25&cursor=7",
|
||
]
|
||
|
||
|
||
def test_security_activity_explains_pending_outcome_confirmation():
|
||
source = SESSION_JS.read_text()
|
||
|
||
assert "Outcome confirmation pending" in source
|
||
assert "event.status === 'pending'" in source
|
||
|
||
|
||
def test_security_activity_labels_consequential_pull_review_decisions():
|
||
source = SESSION_JS.read_text()
|
||
|
||
assert "pull_review_approved: 'Pull request approved'" in source
|
||
assert "pull_review_changes_requested: 'Changes requested'" in source
|
||
|
||
|
||
def test_authenticated_dashboard_load_requests_queued_delivery_resume():
|
||
result = run_session_scenario(
|
||
"""
|
||
await boundary.resumeQueuedWork();
|
||
process.stdout.write(JSON.stringify(state));
|
||
"""
|
||
)
|
||
|
||
assert result["workerMessages"] == [{"type": "stackchain-resume-outbox"}]
|
||
|
||
|
||
@pytest.mark.anyio
|
||
async def test_dashboard_loads_session_boundary_first_and_offers_sign_out():
|
||
html = await dashboard()
|
||
|
||
assert '<script src="static/session.js"></script>' in html
|
||
assert html.index('static/session.js') < html.index('static/markdown.js')
|
||
assert '<button id="sign-out" type="button">Sign out & clear this device</button>' in html
|
||
assert '<button id="sign-out-all" type="button">Sign out all devices</button>' in html
|
||
assert '<button id="active-devices" type="button">Active devices</button>' in html
|
||
assert 'id="active-devices-sheet"' in html
|
||
assert 'aria-label="Active devices and passkeys"' in html
|
||
assert 'aria-labelledby="enrolled-passkeys-title"' in html
|
||
assert 'id="enrolled-passkeys-list"' in html
|
||
assert 'A recovery token will be required after sign-out if you remove this device’s passkey.' in html
|
||
assert 'aria-labelledby="security-activity-title"' in html
|
||
assert 'id="security-activity-list"' in html
|
||
assert '>Load older activity</button>' in html
|
||
assert '#sign-out-all { min-height:44px; }' in html
|