626 lines
24 KiB
Python
626 lines
24 KiB
Python
import json
|
||
import subprocess
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
|
||
from src.views import login
|
||
|
||
|
||
ROOT = Path(__file__).resolve().parents[1]
|
||
LOGIN_JS = ROOT / "frontend" / "login.js"
|
||
|
||
|
||
def test_expired_session_reason_explains_preserved_private_work():
|
||
harness = f"""
|
||
const createLoginController = require({json.dumps(str(LOGIN_JS))});
|
||
const status = {{ textContent: '' }};
|
||
const controller = createLoginController({{
|
||
form: {{ reset: () => {{}} }}, status, button: {{ disabled: false }},
|
||
fetchImpl: async () => new Response('{{}}', {{ status: 200 }}),
|
||
location: {{ replace: () => {{}} }},
|
||
}});
|
||
controller.showReason('session-expired');
|
||
const expired = status.textContent;
|
||
controller.showReason('https://evil.example/redirect');
|
||
process.stdout.write(JSON.stringify({{ expired, ignored: status.textContent }}));
|
||
"""
|
||
result = subprocess.run(
|
||
["node", "-e", harness], text=True, capture_output=True, check=True
|
||
)
|
||
state = json.loads(result.stdout)
|
||
|
||
assert state["expired"] == (
|
||
"Your session expired. Private drafts remain on this device. "
|
||
"Sign in to continue."
|
||
)
|
||
assert state["ignored"] == state["expired"]
|
||
|
||
|
||
def test_idle_session_reason_explains_that_saved_work_will_resume():
|
||
harness = f"""
|
||
const createLoginController = require({json.dumps(str(LOGIN_JS))});
|
||
const status = {{ textContent: '' }};
|
||
const controller = createLoginController({{
|
||
form: {{ reset: () => {{}} }}, status, button: {{ disabled: false }},
|
||
fetchImpl: async () => new Response('{{}}', {{ status: 200 }}),
|
||
location: {{ replace: () => {{}} }},
|
||
}});
|
||
controller.showReason('session-idle');
|
||
process.stdout.write(JSON.stringify({{ status: status.textContent }}));
|
||
"""
|
||
result = subprocess.run(
|
||
["node", "-e", harness], text=True, capture_output=True, check=True
|
||
)
|
||
|
||
assert json.loads(result.stdout)["status"] == (
|
||
"Stackchain locked. Your drafts and queued work are still "
|
||
"on this device. Sign in to resume."
|
||
)
|
||
|
||
|
||
def test_revoked_session_reason_clears_private_data_before_enabling_sign_in():
|
||
harness = f"""
|
||
const createLoginController = require({json.dumps(str(LOGIN_JS))});
|
||
const state = {{cleared:false, snapshots:[]}};
|
||
const status = {{textContent:''}};
|
||
const button = {{disabled:false}};
|
||
const controller = createLoginController({{
|
||
form:{{reset:()=>{{}}}}, status, button,
|
||
fetchImpl:async()=>new Response('{{}}',{{status:200}}),
|
||
location:{{replace:()=>{{}}}},
|
||
clearPrivateDeviceData:async()=>{{
|
||
state.snapshots.push({{status:status.textContent,disabled:button.disabled}});
|
||
await new Promise(resolve=>setTimeout(resolve,10));
|
||
state.cleared=true;
|
||
}},
|
||
}});
|
||
(async()=>{{
|
||
const pending=controller.showReason('session-revoked');
|
||
state.snapshots.push({{status:status.textContent,disabled:button.disabled}});
|
||
await pending;
|
||
state.final={{status:status.textContent,disabled:button.disabled,cleared:state.cleared}};
|
||
process.stdout.write(JSON.stringify(state));
|
||
}})().catch(error=>{{console.error(error);process.exit(1);}});
|
||
"""
|
||
result = subprocess.run(
|
||
["node", "-e", harness], text=True, capture_output=True, check=True
|
||
)
|
||
state = json.loads(result.stdout)
|
||
|
||
assert all(snapshot["disabled"] for snapshot in state["snapshots"])
|
||
assert state["final"] == {
|
||
"status": "This device was remotely signed out. Stackchain private data was cleared. Sign in to use it again.",
|
||
"disabled": False,
|
||
"cleared": True,
|
||
}
|
||
|
||
|
||
@pytest.mark.anyio
|
||
async def test_login_loads_private_data_purger_before_controller():
|
||
html = await login()
|
||
|
||
assert '<script src="static/private-data-registry.js"></script>' in html
|
||
assert '<script src="static/private-device-data.js"></script>' in html
|
||
assert html.index('static/private-data-registry.js') < html.index('static/private-device-data.js')
|
||
assert html.index('static/private-device-data.js') < html.index('static/login.js')
|
||
|
||
|
||
def test_revoked_session_does_not_claim_success_when_purger_is_unavailable():
|
||
harness = f"""
|
||
const createLoginController = require({json.dumps(str(LOGIN_JS))});
|
||
const status={{textContent:''}};
|
||
const button={{disabled:false}};
|
||
const controller=createLoginController({{
|
||
form:{{reset:()=>{{}}}},status,button,
|
||
fetchImpl:async()=>new Response('{{}}',{{status:200}}),location:{{replace:()=>{{}}}},
|
||
}});
|
||
(async()=>{{
|
||
await controller.showReason('session-revoked');
|
||
process.stdout.write(JSON.stringify({{status:status.textContent,disabled:button.disabled}}));
|
||
}})();
|
||
"""
|
||
result = subprocess.run(
|
||
["node", "-e", harness], text=True, capture_output=True, check=True
|
||
)
|
||
|
||
assert json.loads(result.stdout) == {
|
||
"status": "This device was remotely signed out, but private data could not be cleared. Close other Stackchain tabs and clear this site’s data before signing in.",
|
||
"disabled": True,
|
||
}
|
||
|
||
|
||
def test_rate_limited_login_disables_submit_and_counts_down():
|
||
harness = f"""
|
||
const createLoginController = require({json.dumps(str(LOGIN_JS))});
|
||
const state = {{ reset: 0, interval: null }};
|
||
const status = {{ textContent: '' }};
|
||
const button = {{ disabled: false }};
|
||
const form = {{ reset: () => state.reset += 1 }};
|
||
const controller = createLoginController({{
|
||
form, status, button,
|
||
fetchImpl: async () => new Response(JSON.stringify({{ detail: 'Too many sign-in attempts' }}), {{ status: 429, headers: {{ 'Retry-After': '2' }} }}),
|
||
location: {{ replace: () => {{}} }},
|
||
setIntervalImpl: callback => {{ state.interval = callback; return 1; }},
|
||
clearIntervalImpl: () => {{}},
|
||
}});
|
||
(async () => {{
|
||
await controller.submit('never-store-this-token');
|
||
state.initial = {{ disabled: button.disabled, status: status.textContent, reset: state.reset }};
|
||
state.interval();
|
||
state.afterTick = {{ disabled: button.disabled, status: status.textContent }};
|
||
state.interval();
|
||
state.finished = {{ disabled: button.disabled, status: status.textContent }};
|
||
process.stdout.write(JSON.stringify(state));
|
||
}})().catch(error => {{ console.error(error); process.exit(1); }});
|
||
"""
|
||
result = subprocess.run(
|
||
["node", "-e", harness], text=True, capture_output=True, check=True
|
||
)
|
||
state = json.loads(result.stdout)
|
||
|
||
assert state["initial"] == {
|
||
"disabled": True,
|
||
"status": "Too many attempts. Try again in 2 seconds.",
|
||
"reset": 1,
|
||
}
|
||
assert state["afterTick"]["disabled"] is True
|
||
assert state["finished"] == {
|
||
"disabled": False,
|
||
"status": "You can try signing in again.",
|
||
}
|
||
|
||
|
||
def test_stalled_token_login_aborts_once_and_restores_both_sign_in_controls():
|
||
harness = f"""
|
||
const createLoginController = require({json.dumps(str(LOGIN_JS))});
|
||
const state = {{ requests: 0, timeout: null, aborted: false }};
|
||
const status = {{ textContent: '' }};
|
||
const button = {{ disabled: false }};
|
||
const passkeyButton = {{ disabled: false }};
|
||
const controller = createLoginController({{
|
||
form: {{ reset: () => {{}} }}, status, button, passkeyButton,
|
||
credentials: {{ get: async () => {{ throw new Error('unused'); }} }},
|
||
fetchImpl: (_url, options) => {{
|
||
state.requests += 1;
|
||
options.signal.addEventListener('abort', () => state.aborted = true);
|
||
return new Promise((_resolve, reject) => {{
|
||
options.signal.addEventListener('abort', () => reject(options.signal.reason));
|
||
}});
|
||
}},
|
||
location: {{ replace: () => {{}} }},
|
||
requestTimeoutMs: 25,
|
||
setTimeoutImpl: callback => {{ state.timeout = callback; return 7; }},
|
||
clearTimeoutImpl: () => {{}},
|
||
}});
|
||
(async () => {{
|
||
const first = controller.submit('operator-token');
|
||
const duplicate = controller.submit('operator-token');
|
||
state.during = {{ button: button.disabled, passkey: passkeyButton.disabled }};
|
||
state.timeout();
|
||
await Promise.all([first, duplicate]);
|
||
state.after = {{
|
||
button: button.disabled, passkey: passkeyButton.disabled,
|
||
status: status.textContent,
|
||
}};
|
||
process.stdout.write(JSON.stringify(state));
|
||
}})().catch(error => {{ console.error(error); process.exit(1); }});
|
||
"""
|
||
result = subprocess.run(
|
||
["node", "-e", harness], text=True, capture_output=True, check=True
|
||
)
|
||
state = json.loads(result.stdout)
|
||
|
||
assert state["requests"] == 1
|
||
assert state["during"] == {"button": True, "passkey": True}
|
||
assert state["aborted"] is True
|
||
assert state["after"] == {
|
||
"button": False,
|
||
"passkey": False,
|
||
"status": "Sign-in timed out. Check your connection and try again.",
|
||
}
|
||
|
||
|
||
def test_successful_login_resumes_valid_shared_capture_continuation():
|
||
harness = f"""
|
||
const createLoginController = require({json.dumps(str(LOGIN_JS))});
|
||
const state = {{ replaced: null }};
|
||
const controller = createLoginController({{
|
||
form: {{ reset: () => {{}} }}, status: {{ textContent: '' }}, button: {{ disabled: false }},
|
||
fetchImpl: async () => new Response('{{}}', {{ status: 200 }}),
|
||
location: {{ replace: value => state.replaced = value }},
|
||
continuation: './?title=Production+crash&url=https%3A%2F%2Fexample.com',
|
||
}});
|
||
(async () => {{
|
||
await controller.submit('operator-token');
|
||
process.stdout.write(JSON.stringify(state));
|
||
}})().catch(error => {{ console.error(error); process.exit(1); }});
|
||
"""
|
||
result = subprocess.run(
|
||
["node", "-e", harness], text=True, capture_output=True, check=True
|
||
)
|
||
|
||
assert json.loads(result.stdout)["replaced"] == (
|
||
"./?title=Production+crash&url=https%3A%2F%2Fexample.com"
|
||
)
|
||
|
||
|
||
def test_successful_login_resumes_only_valid_search_preview_continuation():
|
||
harness = f"""
|
||
const createLoginController = require({json.dumps(str(LOGIN_JS))});
|
||
async function destination(continuation) {{
|
||
let replaced = null;
|
||
const controller = createLoginController({{
|
||
form: {{ reset: () => {{}} }}, status: {{ textContent: '' }}, button: {{ disabled: false }},
|
||
fetchImpl: async () => new Response('{{}}', {{ status: 200 }}),
|
||
location: {{ replace: value => replaced = value }}, continuation,
|
||
}});
|
||
await controller.submit('operator-token');
|
||
return replaced;
|
||
}}
|
||
(async () => process.stdout.write(JSON.stringify({{
|
||
valid:await destination('./?search=release+blocker&preview=pull%3Astackchain%2Fapi%3A9&search_kind=pull&search_state=open&search_repository=stackchain%2Fapi'),
|
||
malformed:await destination('./?search=release&preview=issue%3Astackchain%2Fapi%3A0'),
|
||
unknown:await destination('./?search=release&next=https%3A%2F%2Fevil.example'),
|
||
}})))().catch(error => {{ console.error(error); process.exit(1); }});
|
||
"""
|
||
result = subprocess.run(["node", "-e", harness], capture_output=True, text=True)
|
||
|
||
assert result.returncode == 0, result.stderr
|
||
assert json.loads(result.stdout) == {
|
||
"valid": "./?search=release+blocker&preview=pull%3Astackchain%2Fapi%3A9&search_kind=pull&search_state=open&search_repository=stackchain%2Fapi",
|
||
"malformed": "./",
|
||
"unknown": "./",
|
||
}
|
||
|
||
|
||
def test_search_preview_login_explains_the_resumable_handoff():
|
||
harness = f"""
|
||
const createLoginController = require({json.dumps(str(LOGIN_JS))});
|
||
const status = {{ textContent: '' }};
|
||
createLoginController({{
|
||
form: {{ reset: () => {{}} }}, status, button: {{ disabled: false }},
|
||
fetchImpl: async () => new Response('{{}}', {{ status: 200 }}),
|
||
location: {{ replace: () => {{}} }},
|
||
continuation: './?search=release&preview=issue%3Astackchain%2Fapi%3A42',
|
||
}});
|
||
process.stdout.write(status.textContent);
|
||
"""
|
||
result = subprocess.run(["node", "-e", harness], capture_output=True, text=True)
|
||
|
||
assert result.returncode == 0, result.stderr
|
||
assert result.stdout == "Sign in to open the shared Search result."
|
||
|
||
|
||
def test_passkey_login_uses_web_authentication_without_sending_the_operator_token():
|
||
harness = f"""
|
||
const createLoginController = require({json.dumps(str(LOGIN_JS))});
|
||
const requests = [];
|
||
const state = {{ replaced: null }};
|
||
const credential = {{
|
||
id:'credential-id', type:'public-key', rawId:Uint8Array.from([1,2,3]).buffer,
|
||
response:{{
|
||
authenticatorData:Uint8Array.from([4]).buffer,
|
||
clientDataJSON:Uint8Array.from([5]).buffer,
|
||
signature:Uint8Array.from([6]).buffer,
|
||
userHandle:null,
|
||
}},
|
||
}};
|
||
const controller = createLoginController({{
|
||
form:{{reset:()=>{{}}}}, status:{{textContent:''}}, button:{{disabled:false}},
|
||
passkeyButton:{{disabled:false}},
|
||
credentials:{{get:async options=>{{ state.publicKey=options.publicKey; return credential; }}}},
|
||
fetchImpl:async (url, options={{}})=>{{
|
||
requests.push({{url, body:options.body ? JSON.parse(options.body) : null}});
|
||
if (url.endsWith('/options')) return new Response(JSON.stringify({{
|
||
challenge:'AQID', rpId:'forge.example', userVerification:'required',
|
||
allowCredentials:[{{type:'public-key',id:'BAUG'}}],
|
||
}}),{{status:200,headers:{{'Content-Type':'application/json'}}}});
|
||
return new Response('{{}}',{{status:200}});
|
||
}},
|
||
location:{{replace:value=>state.replaced=value}},
|
||
}});
|
||
(async()=>{{
|
||
await controller.signInWithPasskey('Timmy’s Pixel');
|
||
process.stdout.write(JSON.stringify({{requests,state}}));
|
||
}})().catch(error=>{{console.error(error);process.exit(1);}});
|
||
"""
|
||
result = subprocess.run(
|
||
["node", "-e", harness], text=True, capture_output=True, check=True
|
||
)
|
||
output = json.loads(result.stdout)
|
||
|
||
assert output["state"]["publicKey"]["challenge"] == {"0": 1, "1": 2, "2": 3}
|
||
assert output["state"]["replaced"] == "./"
|
||
assert output["requests"][1]["url"] == "api/v1/passkeys/authentication/verify"
|
||
assert output["requests"][1]["body"] == {
|
||
"challenge": "AQID",
|
||
"credential": {
|
||
"id": "credential-id",
|
||
"type": "public-key",
|
||
"rawId": "AQID",
|
||
"response": {
|
||
"authenticatorData": "BA",
|
||
"clientDataJSON": "BQ",
|
||
"signature": "Bg",
|
||
"userHandle": None,
|
||
},
|
||
},
|
||
"device_label": "Timmy’s Pixel",
|
||
"action": "sign_in",
|
||
"target": "dashboard",
|
||
}
|
||
assert "access_token" not in json.dumps(output)
|
||
|
||
|
||
def test_stalled_passkey_network_phases_time_out_without_timing_out_biometrics():
|
||
harness = f"""
|
||
const createLoginController = require({json.dumps(str(LOGIN_JS))});
|
||
async function scenario(stalledPhase) {{
|
||
const state = {{ requests: [], timers: [], credentialCalls: 0 }};
|
||
const status = {{textContent:''}};
|
||
const button = {{disabled:false}};
|
||
const passkeyButton = {{disabled:false}};
|
||
const credential = {{
|
||
id:'credential-id', type:'public-key', rawId:Uint8Array.from([1]).buffer,
|
||
response:{{authenticatorData:Uint8Array.from([2]).buffer,
|
||
clientDataJSON:Uint8Array.from([3]).buffer,
|
||
signature:Uint8Array.from([4]).buffer,userHandle:null}},
|
||
}};
|
||
const controller=createLoginController({{
|
||
form:{{reset:()=>{{}}}},status,button,passkeyButton,
|
||
credentials:{{get:async()=>{{state.credentialCalls+=1;return credential;}}}},
|
||
fetchImpl:(url,options)=>{{
|
||
state.requests.push(url);
|
||
const phase=url.endsWith('/options') ? 'options' : 'verify';
|
||
if (phase === stalledPhase) return new Promise((_resolve,reject)=>{{
|
||
options.signal.addEventListener('abort',()=>reject(options.signal.reason));
|
||
}});
|
||
return Promise.resolve(new Response(JSON.stringify({{
|
||
challenge:'AQID',allowCredentials:[]
|
||
}}),{{status:200,headers:{{'Content-Type':'application/json'}}}}));
|
||
}},
|
||
location:{{replace:()=>{{}}}},requestTimeoutMs:25,
|
||
setTimeoutImpl:callback=>{{state.timers.push(callback);return state.timers.length;}},
|
||
clearTimeoutImpl:()=>{{}},
|
||
}});
|
||
const pending=controller.signInWithPasskey('Phone');
|
||
await new Promise(resolve=>setImmediate(resolve));
|
||
if (stalledPhase === 'verify') await new Promise(resolve=>setImmediate(resolve));
|
||
state.timers.at(-1)();
|
||
state.result=await pending;
|
||
state.final={{status:status.textContent,button:button.disabled,passkey:passkeyButton.disabled}};
|
||
return state;
|
||
}}
|
||
(async()=>{{
|
||
process.stdout.write(JSON.stringify({{
|
||
options:await scenario('options'),verify:await scenario('verify'),
|
||
}}));
|
||
}})().catch(error=>{{console.error(error);process.exit(1);}});
|
||
"""
|
||
result = subprocess.run(
|
||
["node", "-e", harness], text=True, capture_output=True, check=True
|
||
)
|
||
state = json.loads(result.stdout)
|
||
|
||
assert state["options"]["credentialCalls"] == 0
|
||
assert state["verify"]["credentialCalls"] == 1
|
||
assert state["verify"]["timers"] == [None, None]
|
||
for phase in ("options", "verify"):
|
||
assert state[phase]["result"] is False
|
||
assert state[phase]["final"] == {
|
||
"status": "Passkey sign-in timed out. Check your connection and try again.",
|
||
"button": False,
|
||
"passkey": False,
|
||
}
|
||
|
||
|
||
def test_login_sends_a_bounded_device_label_with_the_access_token():
|
||
harness = f"""
|
||
const createLoginController = require({json.dumps(str(LOGIN_JS))});
|
||
let request = null;
|
||
const controller = createLoginController({{
|
||
form: {{ reset: () => {{}} }}, status: {{ textContent: '' }}, button: {{ disabled: false }},
|
||
fetchImpl: async (url, options) => {{ request = {{url, body: JSON.parse(options.body)}}; return new Response('{{}}', {{ status: 200 }}); }},
|
||
location: {{ replace: () => {{}} }},
|
||
}});
|
||
(async () => {{
|
||
await controller.submit('operator-token', 'Timmy’s Pixel');
|
||
process.stdout.write(JSON.stringify(request));
|
||
}})().catch(error => {{ console.error(error); process.exit(1); }});
|
||
"""
|
||
result = subprocess.run(
|
||
["node", "-e", harness], text=True, capture_output=True, check=True
|
||
)
|
||
|
||
assert json.loads(result.stdout) == {
|
||
"url": "api/v1/session",
|
||
"body": {"access_token": "operator-token", "device_label": "Timmy’s Pixel"},
|
||
}
|
||
|
||
|
||
def test_shared_capture_login_explains_why_sign_in_is_required():
|
||
harness = f"""
|
||
const createLoginController = require({json.dumps(str(LOGIN_JS))});
|
||
const status = {{ textContent: '' }};
|
||
createLoginController({{
|
||
form: {{ reset: () => {{}} }}, status, button: {{ disabled: false }},
|
||
fetchImpl: async () => new Response('{{}}', {{ status: 200 }}),
|
||
location: {{ replace: () => {{}} }}, continuation: './?title=Shared',
|
||
}});
|
||
process.stdout.write(status.textContent);
|
||
"""
|
||
result = subprocess.run(
|
||
["node", "-e", harness], text=True, capture_output=True, check=True
|
||
)
|
||
|
||
assert result.stdout == "Sign in to continue your shared capture."
|
||
|
||
|
||
def test_login_rejects_untrusted_or_non_share_continuations():
|
||
harness = f"""
|
||
const createLoginController = require({json.dumps(str(LOGIN_JS))});
|
||
async function destination(continuation) {{
|
||
let replaced = null;
|
||
const controller = createLoginController({{
|
||
form: {{ reset: () => {{}} }}, status: {{ textContent: '' }}, button: {{ disabled: false }},
|
||
fetchImpl: async () => new Response('{{}}', {{ status: 200 }}),
|
||
location: {{ replace: value => replaced = value }}, continuation,
|
||
}});
|
||
await controller.submit('operator-token');
|
||
return replaced;
|
||
}}
|
||
(async () => {{
|
||
process.stdout.write(JSON.stringify({{
|
||
crossOrigin: await destination('https://evil.example/steal'),
|
||
otherRoute: await destination('./settings?title=Shared'),
|
||
extraField: await destination('./?title=Shared&next=https%3A%2F%2Fevil.example'),
|
||
prototypeField: await destination('./?toString=Shared'),
|
||
oversized: await destination('./?title=' + 'x'.repeat(201)),
|
||
}}));
|
||
}})().catch(error => {{ console.error(error); process.exit(1); }});
|
||
"""
|
||
result = subprocess.run(
|
||
["node", "-e", harness], text=True, capture_output=True, check=True
|
||
)
|
||
|
||
assert json.loads(result.stdout) == {
|
||
"crossOrigin": "./",
|
||
"otherRoute": "./",
|
||
"extraField": "./",
|
||
"prototypeField": "./",
|
||
"oversized": "./",
|
||
}
|
||
|
||
|
||
def test_login_bootstrap_continues_shared_capture_after_submit():
|
||
harness = f"""
|
||
const fs = require('fs');
|
||
const vm = require('vm');
|
||
const state = {{ replaced: null, submit: null }};
|
||
const elements = {{
|
||
'sign-in': {{ reset: () => {{}}, addEventListener: (_name, handler) => state.submit = handler }},
|
||
'status': {{ textContent: '' }},
|
||
'submit-sign-in': {{ disabled: false }},
|
||
}};
|
||
const context = {{
|
||
URLSearchParams, Response, AbortController, setInterval, clearInterval, setTimeout, clearTimeout,
|
||
FormData: function () {{ return {{ get: () => 'operator-token' }}; }},
|
||
fetch: async () => new Response('{{}}', {{ status: 200 }}),
|
||
document: {{ getElementById: id => elements[id] }},
|
||
window: {{ location: {{
|
||
search: '?continue=.%2F%3Ftitle%3DShared%2Blink%26url%3Dhttps%253A%252F%252Fexample.com',
|
||
replace: value => state.replaced = value,
|
||
}} }},
|
||
}};
|
||
context.fetch.bind = Function.prototype.bind;
|
||
vm.createContext(context);
|
||
vm.runInContext(fs.readFileSync({json.dumps(str(LOGIN_JS))}, 'utf8'), context);
|
||
state.submit({{ preventDefault: () => {{}} }});
|
||
setTimeout(() => process.stdout.write(JSON.stringify({{
|
||
replaced: state.replaced, status: elements.status.textContent,
|
||
}})), 0);
|
||
"""
|
||
result = subprocess.run(
|
||
["node", "-e", harness], text=True, capture_output=True, check=True
|
||
)
|
||
|
||
assert json.loads(result.stdout) == {
|
||
"replaced": "./?title=Shared+link&url=https%3A%2F%2Fexample.com",
|
||
"status": "Signing in…",
|
||
}
|
||
|
||
|
||
def test_login_continues_only_bounded_shared_screenshot_capture():
|
||
harness = f"""
|
||
const createLoginController = require({json.dumps(str(LOGIN_JS))});
|
||
async function destination(continuation) {{
|
||
let replaced = null;
|
||
const controller = createLoginController({{
|
||
form: {{ reset: () => {{}} }}, status: {{ textContent: '' }}, button: {{ disabled: false }},
|
||
fetchImpl: async () => new Response('{{}}', {{ status: 200 }}),
|
||
location: {{ replace: value => replaced = value }}, continuation,
|
||
}});
|
||
await controller.submit('operator-token');
|
||
return replaced;
|
||
}}
|
||
(async () => {{
|
||
process.stdout.write(JSON.stringify({{
|
||
valid: await destination('./?title=Broken&launch=new&shared=image'),
|
||
wrongLaunch: await destination('./?launch=agenda&shared=image'),
|
||
missingLaunch: await destination('./?shared=image'),
|
||
wrongMarker: await destination('./?launch=new&shared=document'),
|
||
duplicateMarker: await destination('./?launch=new&shared=image&shared=image'),
|
||
launchWithoutMarker: await destination('./?launch=new'),
|
||
}}));
|
||
}})().catch(error => {{ console.error(error); process.exit(1); }});
|
||
"""
|
||
result = subprocess.run(
|
||
["node", "-e", harness], text=True, capture_output=True, check=True
|
||
)
|
||
|
||
assert json.loads(result.stdout) == {
|
||
"valid": "./?title=Broken&launch=new&shared=image",
|
||
"wrongLaunch": "./",
|
||
"missingLaunch": "./",
|
||
"wrongMarker": "./",
|
||
"duplicateMarker": "./",
|
||
"launchWithoutMarker": "./?launch=new",
|
||
}
|
||
|
||
|
||
def test_token_and_passkey_login_resume_only_the_private_share_bundle():
|
||
harness = f"""
|
||
const createLoginController = require({json.dumps(str(LOGIN_JS))});
|
||
const credential = {{
|
||
id:'credential-id', type:'public-key', rawId:Uint8Array.from([1]).buffer,
|
||
response:{{authenticatorData:Uint8Array.from([2]).buffer,
|
||
clientDataJSON:Uint8Array.from([3]).buffer,
|
||
signature:Uint8Array.from([4]).buffer,userHandle:null}},
|
||
}};
|
||
async function destination(continuation, authentication) {{
|
||
let replaced = null;
|
||
const controller = createLoginController({{
|
||
form:{{reset:()=>{{}}}}, status:{{textContent:''}}, button:{{disabled:false}},
|
||
passkeyButton:{{disabled:false}}, credentials:{{get:async()=>credential}},
|
||
fetchImpl:async url => url.endsWith('/options')
|
||
? new Response(JSON.stringify({{challenge:'AQID',allowCredentials:[]}}),
|
||
{{status:200,headers:{{'Content-Type':'application/json'}}}})
|
||
: new Response('{{}}',{{status:200}}),
|
||
location:{{replace:value=>replaced=value}}, continuation,
|
||
}});
|
||
if (authentication === 'passkey') await controller.signInWithPasskey('Phone');
|
||
else await controller.submit('operator-token');
|
||
return replaced;
|
||
}}
|
||
(async()=>process.stdout.write(JSON.stringify({{
|
||
token:await destination('./?launch=new&shared=bundle','token'),
|
||
passkey:await destination('./?launch=new&shared=bundle','passkey'),
|
||
extra:await destination('./?launch=new&shared=bundle&title=leak','token'),
|
||
duplicate:await destination('./?launch=new&shared=bundle&shared=bundle','token'),
|
||
wrongLaunch:await destination('./?launch=agenda&shared=bundle','token'),
|
||
}})))().catch(error=>{{console.error(error);process.exit(1);}});
|
||
"""
|
||
result = subprocess.run(
|
||
["node", "-e", harness], text=True, capture_output=True, check=True
|
||
)
|
||
|
||
assert json.loads(result.stdout) == {
|
||
"token": "./?launch=new&shared=bundle",
|
||
"passkey": "./?launch=new&shared=bundle",
|
||
"extra": "./",
|
||
"duplicate": "./",
|
||
"wrongLaunch": "./",
|
||
}
|
||
|
||
|
||
@pytest.mark.anyio
|
||
async def test_login_page_loads_rate_limit_controller():
|
||
html = await login()
|
||
|
||
assert '<script src="static/login.js"></script>' in html
|
||
assert "main{box-sizing:border-box" in html
|
||
assert '<p id="status" role="status" aria-live="polite"></p>' in html
|
||
assert 'name="device_label"' in html
|
||
assert 'maxlength="64"' in html
|