feat: add device passkeys for operator access (Closes #485)
This commit is contained in:
timmy committed
1 parent
131848cc43
commit
ced644b275
15 files changed
+1228
-33
No files matched your search
@@ -39,6 +39,239 @@ async def fresh_grant(client, action: str, target: str) -> str:
|
||||
return response.json()["grant"]
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_passkey_enrollment_options_require_fresh_authorization_and_are_one_time(
|
||||
access_control,
|
||||
):
|
||||
transport = httpx.ASGITransport(app=main.app)
|
||||
async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
|
||||
await client.post(
|
||||
"/api/v1/session",
|
||||
json={
|
||||
"access_token": "correct horse battery staple",
|
||||
"device_label": "Timmy's phone",
|
||||
},
|
||||
)
|
||||
headers = {
|
||||
"Origin": "https://test",
|
||||
"X-CSRF-Token": client.cookies["stackchain_csrf"],
|
||||
}
|
||||
missing = await client.post("/api/v1/passkeys/registration/options", headers=headers)
|
||||
grant = await fresh_grant(client, "enroll_passkey", "current_device")
|
||||
created = await client.post(
|
||||
"/api/v1/passkeys/registration/options",
|
||||
headers={**headers, "X-Step-Up-Grant": grant},
|
||||
)
|
||||
replayed = await client.post(
|
||||
"/api/v1/passkeys/registration/options",
|
||||
headers={**headers, "X-Step-Up-Grant": grant},
|
||||
)
|
||||
|
||||
assert missing.status_code == 428
|
||||
assert missing.json()["detail"]["action"] == "enroll_passkey"
|
||||
assert created.status_code == 201
|
||||
assert created.headers["cache-control"] == "no-store"
|
||||
options = created.json()
|
||||
assert options["rp"] == {"id": "test", "name": "Stackchain Dashboard"}
|
||||
assert options["user"]["name"] == "stackchain-operator"
|
||||
assert options["authenticatorSelection"]["userVerification"] == "required"
|
||||
assert options["challenge"]
|
||||
assert replayed.status_code == 428
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_enrolled_passkey_can_sign_in_without_the_operator_token(
|
||||
access_control, monkeypatch
|
||||
):
|
||||
class VerifiedRegistration:
|
||||
credential_id = b"phone-credential"
|
||||
credential_public_key = b"credential-public-key"
|
||||
sign_count = 0
|
||||
|
||||
class VerifiedAuthentication:
|
||||
new_sign_count = 1
|
||||
|
||||
monkeypatch.setattr(
|
||||
main.passkeys,
|
||||
"verify_registration",
|
||||
lambda **_kwargs: VerifiedRegistration(),
|
||||
raising=False,
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
main.passkeys,
|
||||
"verify_authentication",
|
||||
lambda **_kwargs: VerifiedAuthentication(),
|
||||
raising=False,
|
||||
)
|
||||
transport = httpx.ASGITransport(app=main.app)
|
||||
async with httpx.AsyncClient(transport=transport, base_url="https://test") as bootstrap:
|
||||
await bootstrap.post(
|
||||
"/api/v1/session",
|
||||
json={"access_token": "correct horse battery staple", "device_label": "Phone"},
|
||||
)
|
||||
csrf_headers = {
|
||||
"Origin": "https://test",
|
||||
"X-CSRF-Token": bootstrap.cookies["stackchain_csrf"],
|
||||
}
|
||||
grant = await fresh_grant(bootstrap, "enroll_passkey", "current_device")
|
||||
options = await bootstrap.post(
|
||||
"/api/v1/passkeys/registration/options",
|
||||
headers={**csrf_headers, "X-Step-Up-Grant": grant},
|
||||
)
|
||||
enrolled = await bootstrap.post(
|
||||
"/api/v1/passkeys/registration/verify",
|
||||
json={"challenge": options.json()["challenge"], "credential": {"id": "fake"}},
|
||||
headers=csrf_headers,
|
||||
)
|
||||
|
||||
async with httpx.AsyncClient(transport=transport, base_url="https://test") as returning:
|
||||
sign_in_options = await returning.post("/api/v1/passkeys/authentication/options")
|
||||
signed_in = await returning.post(
|
||||
"/api/v1/passkeys/authentication/verify",
|
||||
json={
|
||||
"challenge": sign_in_options.json()["challenge"],
|
||||
"credential": {"id": "cGhvbmUtY3JlZGVudGlhbA"},
|
||||
"device_label": "Phone",
|
||||
"action": "sign_in",
|
||||
"target": "dashboard",
|
||||
},
|
||||
)
|
||||
|
||||
assert enrolled.status_code == 201
|
||||
assert enrolled.json() == {"enrolled": True}
|
||||
assert sign_in_options.status_code == 200
|
||||
assert sign_in_options.json()["allowCredentials"][0]["id"] == "cGhvbmUtY3JlZGVudGlhbA"
|
||||
assert signed_in.status_code == 200
|
||||
assert signed_in.json() == {"authenticated": True, "method": "passkey"}
|
||||
assert "stackchain_session=" in signed_in.headers["set-cookie"]
|
||||
assert "correct horse battery staple" not in signed_in.text
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_passkey_fresh_authorization_is_exact_target_bound_and_single_use(
|
||||
access_control, monkeypatch
|
||||
):
|
||||
class VerifiedRegistration:
|
||||
credential_id = b"phone-credential"
|
||||
credential_public_key = b"credential-public-key"
|
||||
sign_count = 0
|
||||
|
||||
class VerifiedAuthentication:
|
||||
new_sign_count = 1
|
||||
|
||||
monkeypatch.setattr(
|
||||
main.passkeys, "verify_registration", lambda **_kwargs: VerifiedRegistration()
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
main.passkeys, "verify_authentication", lambda **_kwargs: VerifiedAuthentication()
|
||||
)
|
||||
transport = httpx.ASGITransport(app=main.app)
|
||||
async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
|
||||
await client.post(
|
||||
"/api/v1/session",
|
||||
json={"access_token": "correct horse battery staple", "device_label": "Phone"},
|
||||
)
|
||||
headers = {
|
||||
"Origin": "https://test",
|
||||
"X-CSRF-Token": client.cookies["stackchain_csrf"],
|
||||
}
|
||||
enrollment_grant = await fresh_grant(client, "enroll_passkey", "current_device")
|
||||
registration = await client.post(
|
||||
"/api/v1/passkeys/registration/options",
|
||||
headers={**headers, "X-Step-Up-Grant": enrollment_grant},
|
||||
)
|
||||
await client.post(
|
||||
"/api/v1/passkeys/registration/verify",
|
||||
json={"challenge": registration.json()["challenge"], "credential": {"id": "fake"}},
|
||||
headers=headers,
|
||||
)
|
||||
options = await client.post(
|
||||
"/api/v1/passkeys/authorization/options",
|
||||
json={"action": "close_issue", "target": "stackchain/api#7"},
|
||||
headers=headers,
|
||||
)
|
||||
ceremony = {
|
||||
"challenge": options.json()["challenge"],
|
||||
"credential": {"id": "cGhvbmUtY3JlZGVudGlhbA"},
|
||||
"action": "close_issue",
|
||||
"target": "stackchain/api#7",
|
||||
}
|
||||
authorized = await client.post(
|
||||
"/api/v1/passkeys/authorization/verify", json=ceremony, headers=headers
|
||||
)
|
||||
replayed = await client.post(
|
||||
"/api/v1/passkeys/authorization/verify", json=ceremony, headers=headers
|
||||
)
|
||||
wrong_target = await client.post(
|
||||
"/api/v1/passkeys/authorization/verify",
|
||||
json={**ceremony, "target": "stackchain/api#8"},
|
||||
headers=headers,
|
||||
)
|
||||
|
||||
assert options.status_code == 200
|
||||
assert authorized.status_code == 201
|
||||
assert authorized.json()["grant"]
|
||||
assert authorized.json()["expires_in"] == 90
|
||||
assert replayed.status_code == 409
|
||||
assert wrong_target.status_code == 409
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_revoking_an_enrolled_device_also_revokes_its_passkey(
|
||||
access_control, monkeypatch
|
||||
):
|
||||
class VerifiedRegistration:
|
||||
credential_id = b"phone-credential"
|
||||
credential_public_key = b"credential-public-key"
|
||||
sign_count = 0
|
||||
|
||||
monkeypatch.setattr(
|
||||
main.passkeys, "verify_registration", lambda **_kwargs: VerifiedRegistration()
|
||||
)
|
||||
transport = httpx.ASGITransport(app=main.app)
|
||||
async with (
|
||||
httpx.AsyncClient(transport=transport, base_url="https://test") as phone,
|
||||
httpx.AsyncClient(transport=transport, base_url="https://test") as laptop,
|
||||
):
|
||||
await phone.post(
|
||||
"/api/v1/session",
|
||||
json={"access_token": "correct horse battery staple", "device_label": "Phone"},
|
||||
)
|
||||
phone_headers = {
|
||||
"Origin": "https://test",
|
||||
"X-CSRF-Token": phone.cookies["stackchain_csrf"],
|
||||
}
|
||||
enrollment_grant = await fresh_grant(phone, "enroll_passkey", "current_device")
|
||||
registration = await phone.post(
|
||||
"/api/v1/passkeys/registration/options",
|
||||
headers={**phone_headers, "X-Step-Up-Grant": enrollment_grant},
|
||||
)
|
||||
await phone.post(
|
||||
"/api/v1/passkeys/registration/verify",
|
||||
json={"challenge": registration.json()["challenge"], "credential": {"id": "fake"}},
|
||||
headers=phone_headers,
|
||||
)
|
||||
await laptop.post(
|
||||
"/api/v1/session",
|
||||
json={"access_token": "correct horse battery staple", "device_label": "Laptop"},
|
||||
)
|
||||
devices = (await laptop.get("/api/v1/sessions")).json()["devices"]
|
||||
phone_device = next(item for item in devices if item["device_label"] == "Phone")
|
||||
grant = await fresh_grant(laptop, "revoke_device", phone_device["management_id"])
|
||||
revoked = await laptop.delete(
|
||||
f"/api/v1/sessions/{phone_device['management_id']}",
|
||||
headers={
|
||||
"Origin": "https://test",
|
||||
"X-CSRF-Token": laptop.cookies["stackchain_csrf"],
|
||||
"X-Step-Up-Grant": grant,
|
||||
},
|
||||
)
|
||||
passkey_options = await laptop.post("/api/v1/passkeys/authentication/options")
|
||||
|
||||
assert revoked.status_code == 200
|
||||
assert passkey_options.status_code == 404
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_default_operator_mode_fails_closed_before_gitea_when_secrets_are_missing(monkeypatch):
|
||||
monkeypatch.delenv("STACKCHAIN_DASHBOARD_AUTH_MODE", raising=False)
|
||||
|
||||
@@ -297,6 +297,77 @@ process.stdout.write(JSON.stringify(state));
|
||||
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(
|
||||
"""
|
||||
|
||||
@@ -193,6 +193,67 @@ const controller = createLoginController({{
|
||||
)
|
||||
|
||||
|
||||
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_login_sends_a_bounded_device_label_with_the_access_token():
|
||||
harness = f"""
|
||||
const createLoginController = require({json.dumps(str(LOGIN_JS))});
|
||||
|
||||
Reference in new issue
Block a user