morgan/virtual-webauthn
1
0
Fork 0
mirror of https://github.com/morgan9e/virtual-webauthn synced 2026-04-14 00:04:09 +09:00
Code Issues Packages Projects Releases Wiki Activity

Rewrite in Rust, refine extension

Browse Source
This commit is contained in:
Morgan
2026-03-30 22:57:08 +09:00
parent b3ba21129c
commit 6333b26f57
12 changed files with 1083 additions and 728 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
.gitignore vendored
View File

@@ -1,3 +1,2 @@
passkey.json
__pycache__/
dist/
target/
Cargo.lock

23
Cargo.toml Normal file
View File

@@ -0,0 +1,23 @@
[package]
name = "virtual-webauthn"
version = "0.1.0"
edition = "2024"
[dependencies]
serde = { version = "1", features = ["derive"] }
serde_json = "1"
p256 = { version = "0.13", features = ["ecdsa", "pem"] }
ecdsa = { version = "0.16", features = ["signing", "der"] }
aes-gcm = "0.10"
scrypt = "0.11"
sha2 = "0.10"
ciborium = "0.2"
base64ct = { version = "1", features = ["std"] }
rand = "0.8"
log = "0.4"
env_logger = "0.11"
dirs = "6"
[profile.release]
lto = true
strip = true

38
Makefile
View File

@@ -1,29 +1,21 @@
MODE ?= virtual
NMH_DIR ?= $(HOME)/.librewolf/native-messaging-hosts
BIN_DIR ?= $(HOME)/.librewolf/external_application
EXT_ID ?= com.example.virtual_webauthn
.PHONY: build chrome firefox clean run run-physical install
.PHONY: build clean install extension
build: chrome firefox
build:
cargo build --release
chrome: dist/chrome
firefox: dist/virtual-webauthn.xpi
dist/chrome: extension/*
@rm -rf $@
@mkdir -p $@
cp extension/* $@/
dist/virtual-webauthn.xpi: extension/*
@mkdir -p dist
cd extension && zip -r ../$@ . -x '.*'
extension:
@mkdir -p target
cd extension && zip -r ../target/virtual-webauthn.xpi . -x '.*'
clean:
rm -rf dist/
cargo clean
run:
cd server && python main.py --mode $(MODE)
run-physical:
cd server && python main.py --mode physical
install:
pip install -r requirements.txt
install: build
@mkdir -p $(BIN_DIR) $(NMH_DIR)
install -m755 target/release/virtual-webauthn $(BIN_DIR)/virtual-webauthn
cp virtual_webauthn.json $(NMH_DIR)/$(EXT_ID).json
@sed -i "s,/PLACEHOLDER,$(BIN_DIR)," $(NMH_DIR)/$(EXT_ID).json

142
extension/background.js
View File

@@ -1,50 +1,124 @@
const API_URL = "http://127.0.0.1:20492";
const HOST_NAME = "com.example.virtual_webauthn";
async function apiFetch(method, path, body) {
const opts = { method, headers: {} };
if (body !== undefined) {
opts.headers["Content-Type"] = "application/json";
opts.body = JSON.stringify(body);
}
const response = await fetch(API_URL + path, opts);
if (!response.ok) {
const detail = await response.json().catch(() => ({}));
throw new Error(detail.detail || `Server error: ${response.status}`);
}
return response.json();
let port = null;
let seq = 0;
const pending = new Map();
let sessionKey = null;
function connect() {
if (port) return;
try {
port = chrome.runtime.connectNative(HOST_NAME);
} catch {
return;
}
// --- Icon status polling ---
port.onMessage.addListener((msg) => {
if (msg.sessionKey) {
sessionKey = msg.sessionKey;
}
const cb = pending.get(msg.id);
if (cb) {
pending.delete(msg.id);
cb(msg);
}
});
port.onDisconnect.addListener(() => {
port = null;
for (const [id, cb] of pending) {
cb({ id, success: false, error: "Host disconnected" });
}
pending.clear();
updateIcon(false);
});
updateIcon(true);
}
function sendNative(msg) {
return new Promise((resolve, reject) => {
connect();
if (!port) {
reject(new Error("Cannot connect to native host"));
return;
}
const id = ++seq;
const timer = setTimeout(() => {
pending.delete(id);
reject(new Error("Timed out"));
}, 120_000);
pending.set(id, (resp) => {
clearTimeout(timer);
resolve(resp);
});
port.postMessage({ ...msg, id });
});
}
// --- Icon status ---
let lastStatus = null;
async function updateIcon() {
function updateIcon(connected) {
const status = connected ? "ok" : "err";
if (status === lastStatus) return;
lastStatus = status;
const icon = connected ? "icon-green.svg" : "icon-red.svg";
const title = connected
? "Virtual WebAuthn — Connected"
: "Virtual WebAuthn — Disconnected";
chrome.action.setIcon({ path: icon });
chrome.action.setTitle({ title });
}
async function pingLoop() {
try {
await apiFetch("GET", "/ping");
if (lastStatus !== "ok") {
chrome.action.setIcon({ path: "icon-green.svg" });
chrome.action.setTitle({ title: "Virtual WebAuthn — Connected" });
lastStatus = "ok";
}
const resp = await sendNative({ type: "ping" });
updateIcon(resp.success === true);
} catch {
if (lastStatus !== "err") {
chrome.action.setIcon({ path: "icon-red.svg" });
chrome.action.setTitle({ title: "Virtual WebAuthn — Disconnected" });
lastStatus = "err";
}
updateIcon(false);
}
setTimeout(pingLoop, 10_000);
}
updateIcon();
setInterval(updateIcon, 5000);
pingLoop();
// --- Message relay ---
// --- Message relay from content script ---
chrome.runtime.onMessage.addListener((message, sender, sendResponse) => {
if (message.type === "VWEBAUTHN_REQUEST") {
apiFetch("POST", "", { type: message.action, data: message.payload })
.then((data) => sendResponse({ success: true, data }))
.catch((error) => sendResponse({ success: false, error: error.message }));
return true;
if (message.type !== "VWEBAUTHN_REQUEST") return;
const msg = {
type: message.action,
data: message.payload,
};
// Password from content.js (already in isolated context)
if (message.password) {
msg.password = message.password;
} else if (sessionKey) {
msg.sessionKey = sessionKey;
}
if (message.action === "list" && message.rpId) {
msg.rpId = message.rpId;
}
sendNative(msg)
.then((resp) => {
if (!resp.success) {
const err = resp.error || "";
if (err.includes("session") || err.includes("Session")) {
sessionKey = null;
}
}
sendResponse(resp);
})
.catch((error) => {
sendResponse({ success: false, error: error.message });
});
return true;
});

138
extension/content.js
View File

@@ -3,14 +3,146 @@ s.src = chrome.runtime.getURL("inject.js");
s.onload = () => s.remove();
(document.documentElement || document.head).appendChild(s);
// --- Password prompt (closed shadow DOM, isolated context) ---
function showPasswordPrompt(title, needsConfirm) {
return new Promise((resolve) => {
const host = document.createElement("div");
host.style.cssText = "position:fixed;top:0;left:0;width:100%;height:100%;z-index:2147483647";
const shadow = host.attachShadow({ mode: "closed" });
shadow.innerHTML = `
<style>
.overlay { position:fixed;top:0;left:0;width:100%;height:100%;background:rgba(0,0,0,.3); }
.popup {
position:fixed;top:50%;left:50%;transform:translate(-50%,-50%);
background:#fff;color:#000;border:1px solid #bbb;border-radius:8px;padding:16px;
max-width:320px;box-shadow:0 4px 16px rgba(0,0,0,.18);
font-family:system-ui,-apple-system,sans-serif;font-size:14px;line-height:1.4;
}
.title { margin:0 0 12px;font-size:15px;font-weight:600; }
input {
width:100%;padding:8px 10px;border:1px solid #ccc;border-radius:4px;
font-size:14px;margin-bottom:8px;box-sizing:border-box;
}
.err { color:#dc2626;font-size:12px;margin-bottom:8px;display:none; }
.btns { display:flex;gap:8px;justify-content:flex-end; }
button {
padding:8px 16px;border:none;border-radius:4px;font-size:13px;cursor:pointer;font-weight:500;
}
.cancel { background:#f0f0f0;color:#333; }
.ok { background:#222;color:#fff; }
</style>
<div class="overlay"></div>
<div class="popup">
<div class="title"></div>
<input type="password" class="pw" placeholder="Password">
${needsConfirm ? '<input type="password" class="pw2" placeholder="Confirm password">' : ""}
<div class="err"></div>
<div class="btns">
<button class="cancel">Cancel</button>
<button class="ok">Unlock</button>
</div>
</div>`;
const cleanup = () => host.remove();
shadow.querySelector(".title").textContent = title;
const pw = shadow.querySelector(".pw");
const pw2 = shadow.querySelector(".pw2");
const errEl = shadow.querySelector(".err");
const submit = () => {
if (!pw.value) {
errEl.textContent = "Password required";
errEl.style.display = "";
return;
}
if (needsConfirm && pw.value !== pw2.value) {
errEl.textContent = "Passwords do not match";
errEl.style.display = "";
return;
}
const val = pw.value;
cleanup();
resolve(val);
};
shadow.querySelector(".ok").onclick = submit;
shadow.querySelector(".cancel").onclick = () => { cleanup(); resolve(null); };
shadow.querySelector(".overlay").onclick = () => { cleanup(); resolve(null); };
const onKey = (e) => { if (e.key === "Enter") submit(); };
pw.addEventListener("keydown", onKey);
if (pw2) pw2.addEventListener("keydown", onKey);
document.body.appendChild(host);
pw.focus();
});
}
// --- Message relay with auth handling ---
async function sendToHost(msg) {
const response = await chrome.runtime.sendMessage(msg);
if (chrome.runtime.lastError) throw new Error(chrome.runtime.lastError.message);
return response;
}
async function handleRequest(action, payload, rpId) {
const msg = { type: "VWEBAUTHN_REQUEST", action, payload };
if (rpId) msg.rpId = rpId;
// No-auth actions pass through directly
if (action === "list" || action === "status" || action === "ping") {
return sendToHost(msg);
}
// Try with session first (no password)
let response = await sendToHost(msg);
// If session worked, done
if (response.success) return response;
// Need password — check if first-time setup
const isSessionError = response.error?.includes("session") || response.error?.includes("Session")
|| response.error?.includes("Password or session");
if (!isSessionError) return response; // real error, don't retry
let statusResp;
try {
statusResp = await sendToHost({ type: "VWEBAUTHN_REQUEST", action: "status", payload: {} });
} catch {
return response;
}
const needsSetup = statusResp.success && statusResp.data?.needsSetup;
const title = needsSetup
? "Virtual WebAuthn — Set Password"
: `Virtual WebAuthn — ${action === "create" ? "Create Credential" : "Authenticate"}`;
// Retry loop — allow 3 password attempts
for (let attempt = 0; attempt < 3; attempt++) {
const password = await showPasswordPrompt(
attempt > 0 ? "Wrong password — try again" : title,
needsSetup,
);
if (!password) return { success: false, error: "Password prompt cancelled" };
msg.password = password;
const retry = await sendToHost(msg);
if (retry.success || !retry.error?.includes("password")) return retry;
}
return { success: false, error: "Too many failed attempts" };
}
window.addEventListener("message", async (event) => {
if (event.source !== window || event.data?.type !== "VWEBAUTHN_REQUEST") return;
const { id, action, payload } = event.data;
try {
const response = await chrome.runtime.sendMessage({
type: "VWEBAUTHN_REQUEST", action, payload,
});
const response = await handleRequest(action, payload, event.data.rpId);
window.postMessage({ type: "VWEBAUTHN_RESPONSE", id, ...response }, "*");
} catch (error) {
window.postMessage({ type: "VWEBAUTHN_RESPONSE", id, success: false, error: error.message }, "*");

138
extension/inject.js
View File

@@ -18,33 +18,20 @@
return bytes.buffer;
}
const STYLE = {
popup: {
// --- UI (toast + credential selector only, no password) ---
const POPUP_STYLE = {
position: "fixed", top: "20px", right: "20px",
background: "#fff", color: "#000", border: "1px solid #bbb",
borderRadius: "8px", padding: "16px", zIndex: "2147483647",
maxWidth: "320px", boxShadow: "0 4px 16px rgba(0,0,0,.18)",
fontFamily: "system-ui, -apple-system, sans-serif",
fontSize: "14px", lineHeight: "1.4",
},
title: {
margin: "0 0 12px", fontSize: "15px", fontWeight: "600",
},
option: {
padding: "10px 12px", cursor: "pointer", borderRadius: "6px",
transition: "background .1s",
},
};
function createPopup() {
const el = document.createElement("div");
Object.assign(el.style, STYLE.popup);
return el;
}
function showToast(message) {
const toast = createPopup();
Object.assign(toast.style, { padding: "12px 16px", cursor: "default" });
const toast = document.createElement("div");
Object.assign(toast.style, { ...POPUP_STYLE, padding: "12px 16px", cursor: "default" });
toast.innerHTML =
`<div style="display:flex;align-items:center;gap:8px">` +
`<svg width="18" height="18" viewBox="0 0 24 24" fill="none" stroke="#555" stroke-width="2">` +
@@ -57,19 +44,22 @@
function showCredentialSelector(credentials) {
return new Promise((resolve) => {
const popup = createPopup();
const popup = document.createElement("div");
Object.assign(popup.style, POPUP_STYLE);
const title = document.createElement("div");
title.textContent = "Select a passkey";
Object.assign(title.style, STYLE.title);
Object.assign(title.style, { margin: "0 0 12px", fontSize: "15px", fontWeight: "600" });
popup.appendChild(title);
const optStyle = { padding: "10px 12px", cursor: "pointer", borderRadius: "6px", transition: "background .1s" };
credentials.forEach((cred) => {
const opt = document.createElement("div");
Object.assign(opt.style, STYLE.option);
Object.assign(opt.style, optStyle);
const date = new Date(cred.created * 1000).toLocaleString();
opt.innerHTML =
`<strong>${cred.username || "Unknown"}</strong>` +
`<strong>${cred.user_name || "Unknown"}</strong>` +
`<div style="font-size:.8em;color:#666;margin-top:2px">${date}</div>`;
opt.onmouseover = () => (opt.style.background = "#f0f0f0");
opt.onmouseout = () => (opt.style.background = "transparent");
@@ -79,7 +69,7 @@
const cancel = document.createElement("div");
Object.assign(cancel.style, {
...STYLE.option, textAlign: "center", color: "#888",
...optStyle, textAlign: "center", color: "#888",
marginTop: "4px", borderTop: "1px solid #eee", paddingTop: "10px",
});
cancel.textContent = "Cancel";
@@ -92,6 +82,8 @@
});
}
// --- Messaging (no password in postMessage) ---
const pending = new Map();
let seq = 0;
@@ -121,20 +113,9 @@
});
}
navigator.credentials.create = async function (options) {
const toast = showToast("Waiting for passkey...");
try {
const pk = options.publicKey;
const resp = await request("create", {
publicKey: {
...pk,
challenge: toB64url(pk.challenge),
user: { ...pk.user, id: toB64url(pk.user.id) },
excludeCredentials: pk.excludeCredentials?.map((c) => ({ ...c, id: toB64url(c.id) })),
},
origin: location.origin,
});
// --- Response builders ---
function buildCreateResponse(resp) {
return {
id: resp.id,
type: resp.type,
@@ -150,34 +131,9 @@
},
getClientExtensionResults: () => ({}),
};
} catch (err) {
console.warn("[VirtualWebAuthn] create fallback:", err.message);
return origCreate(options);
} finally {
toast.remove();
}
};
navigator.credentials.get = async function (options) {
const toast = showToast("Waiting for passkey...");
try {
const pk = options.publicKey;
let resp = await request("get", {
publicKey: {
...pk,
challenge: toB64url(pk.challenge),
allowCredentials: pk.allowCredentials?.map((c) => ({ ...c, id: toB64url(c.id) })),
},
origin: location.origin,
});
toast.remove();
if (Array.isArray(resp)) {
resp = await showCredentialSelector(resp);
if (!resp) throw new Error("User cancelled");
}
function buildGetResponse(resp) {
const cred = {
id: resp.id,
type: resp.type,
@@ -194,6 +150,64 @@
cred.response.userHandle = fromB64url(resp.response.userHandle);
}
return cred;
}
// --- WebAuthn overrides ---
navigator.credentials.create = async function (options) {
const toast = showToast("Creating passkey...");
try {
const pk = options.publicKey;
const resp = await request("create", {
publicKey: {
...pk,
challenge: toB64url(pk.challenge),
user: { ...pk.user, id: toB64url(pk.user.id) },
excludeCredentials: pk.excludeCredentials?.map((c) => ({ ...c, id: toB64url(c.id) })),
},
origin: location.origin,
});
return buildCreateResponse(resp);
} catch (err) {
console.warn("[VirtualWebAuthn] create fallback:", err.message);
return origCreate(options);
} finally {
toast.remove();
}
};
navigator.credentials.get = async function (options) {
const pk = options.publicKey;
// Check if we have credentials for this rpId (no auth needed)
try {
const creds = await request("list", { rpId: pk.rpId || "" });
if (Array.isArray(creds) && creds.length === 0) {
return origGet(options);
}
} catch {
return origGet(options);
}
const toast = showToast("Authenticating...");
try {
let resp = await request("get", {
publicKey: {
...pk,
challenge: toB64url(pk.challenge),
allowCredentials: pk.allowCredentials?.map((c) => ({ ...c, id: toB64url(c.id) })),
},
origin: location.origin,
});
toast.remove();
if (Array.isArray(resp)) {
resp = await showCredentialSelector(resp);
if (!resp) throw new Error("User cancelled");
}
return buildGetResponse(resp);
} catch (err) {
console.warn("[VirtualWebAuthn] get fallback:", err.message);
return origGet(options);

9
extension/manifest.json
View File

@@ -3,9 +3,14 @@
"name": "Virtual WebAuthn",
"version": "1.0",
"description": "Not your keys, not your credential",
"host_permissions": [
"http://127.0.0.1:20492/*"
"permissions": [
"nativeMessaging"
],
"browser_specific_settings": {
"gecko": {
"id": "virtual-webauthn@local"
}
},
"action": {
"default_icon": "icon-red.svg",
"default_title": "Virtual WebAuthn — Disconnected"

5
requirements.txt
View File

@@ -1,5 +0,0 @@
cbor2
pycryptodome
fido2
fastapi
uvicorn

147
server/main.py
View File

@@ -1,147 +0,0 @@
import argparse
import logging
import traceback
from typing import Dict, Any
from fastapi import FastAPI, HTTPException, Request
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
import uvicorn
from passkey import VirtualPasskey, PhysicalPasskey, _AuthError, _b64url_decode
log = logging.getLogger("vwebauthn")
app = FastAPI(title="Virtual WebAuthn")
passkey_cls = VirtualPasskey
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_methods=["GET", "POST", "DELETE"],
allow_headers=["Content-Type"],
)
class WebAuthnRequest(BaseModel):
type: str
data: Dict[str, Any]
@app.post("/")
async def handle(req: WebAuthnRequest):
webauthn = passkey_cls()
options = req.data.get("publicKey", {})
origin = req.data.get("origin", "")
log.info("POST / type=%s origin=%s", req.type, origin)
rp = options.get("rp", {}).get("id") or options.get("rpId", "")
if rp:
log.info(" rp_id=%s", rp)
try:
if req.type == "create":
user = options.get("user", {})
log.info(" create user=%s", user.get("displayName") or user.get("name", "?"))
result = webauthn.create(options, origin)
log.info(" created credential id=%s", result.get("id", "?")[:16] + "...")
return result
elif req.type == "get":
allowed = options.get("allowCredentials", [])
log.info(" get allowCredentials=%d", len(allowed))
result = webauthn.get(options, origin)
log.info(" authenticated credential id=%s counter=%s",
result.get("id", "?")[:16] + "...",
result.get("response", {}).get("authenticatorData", "?"))
return result
else:
raise HTTPException(status_code=400, detail=f"Unknown type: {req.type}")
except HTTPException:
raise
except _AuthError as e:
log.warning(" auth error: %s", e)
raise HTTPException(status_code=401, detail=str(e))
except (VirtualPasskey.CredNotFoundError, VirtualPasskey.InputDataError,
PhysicalPasskey.InputDataError) as e:
log.warning(" client error: %s", e)
raise HTTPException(status_code=400, detail=str(e))
except Exception as e:
log.error(" unhandled error: %s", e, exc_info=True)
raise HTTPException(status_code=500, detail=str(e))
@app.get("/ping")
def ping():
mode = "physical" if passkey_cls is PhysicalPasskey else "virtual"
log.debug("GET /ping mode=%s", mode)
return {"status": "ok", "mode": mode}
@app.get("/credentials")
def list_credentials():
log.info("GET /credentials")
if passkey_cls is PhysicalPasskey:
raise HTTPException(status_code=400, detail="Not available in physical mode")
webauthn = VirtualPasskey()
try:
password = webauthn._ask_password("Virtual WebAuthn — List Credentials")
creds = webauthn._load_credentials(password)
except _AuthError as e:
log.warning(" auth error: %s", e)
raise HTTPException(status_code=401, detail=str(e))
log.info(" loaded %d credentials", len(creds))
return [
{
"id": cid,
"rp_id": _b64url_decode(c["rp_id"]).decode("utf-8", errors="ignore"),
"user_name": c.get("user_name", ""),
"created": c.get("created", 0),
"counter": c.get("counter", 0),
}
for cid, c in creds.items()
]
@app.delete("/credentials/{credential_id}")
def delete_credential(credential_id: str):
log.info("DELETE /credentials/%s", credential_id[:16] + "...")
if passkey_cls is PhysicalPasskey:
raise HTTPException(status_code=400, detail="Not available in physical mode")
webauthn = VirtualPasskey()
try:
password = webauthn._ask_password("Virtual WebAuthn — Delete Credential")
webauthn.credentials = webauthn._load_credentials(password)
except _AuthError as e:
log.warning(" auth error: %s", e)
raise HTTPException(status_code=401, detail=str(e))
if credential_id not in webauthn.credentials:
log.warning(" credential not found")
raise HTTPException(status_code=404, detail="Credential not found")
del webauthn.credentials[credential_id]
webauthn._save_credentials(password)
log.info(" deleted successfully")
return {"status": "deleted"}
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Virtual WebAuthn Server")
parser.add_argument(
"--mode", choices=["virtual", "physical"], default="virtual",
help="Passkey mode: virtual (software keys) or physical (USB FIDO2 device)"
)
parser.add_argument("--host", default="127.0.0.1")
parser.add_argument("--port", type=int, default=20492)
parser.add_argument("-v", "--verbose", action="store_true", help="Verbose logging")
args = parser.parse_args()
level = logging.DEBUG if args.verbose else logging.INFO
logging.basicConfig(
level=level,
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
datefmt="%H:%M:%S",
)
if args.mode == "physical":
passkey_cls = PhysicalPasskey
else:
passkey_cls = VirtualPasskey
log.info("Mode: %s", args.mode)
uvicorn.run(app, host=args.host, port=args.port, log_level="debug" if args.verbose else "info")

445
server/passkey.py
View File

@@ -1,445 +0,0 @@
import json
import base64
import logging
import os
import time
import hashlib
import struct
import subprocess
import cbor2
from Crypto.PublicKey import ECC
from Crypto.Signature import DSS
from Crypto.Hash import SHA256
from Crypto.Cipher import AES
from typing import Dict, Any, Optional
log = logging.getLogger("vwebauthn.passkey")
ZENITY_BINARY = os.environ.get("ZENITY_BINARY", "zenity")
def _b64url_encode(data: bytes) -> str:
return base64.urlsafe_b64encode(data).decode().rstrip('=')
def _b64url_decode(data: str) -> bytes:
return base64.urlsafe_b64decode(data + "===")
def _zenity(args: list, timeout: int = 120) -> str:
try:
result = subprocess.run(
[ZENITY_BINARY] + args,
capture_output=True, text=True, timeout=timeout
)
except FileNotFoundError:
raise RuntimeError(f"{ZENITY_BINARY} is not installed")
if result.returncode != 0:
return None
return result.stdout.strip()
def _zenity_password(title: str) -> str:
pw = _zenity(["--password", "--title", title])
if pw is None:
raise _AuthError("Password prompt cancelled")
if not pw:
raise _AuthError("Empty password")
return pw
def _zenity_entry(title: str, text: str, hide: bool = False) -> str:
args = ["--entry", "--title", title, "--text", text]
if hide:
args.append("--hide-text")
return _zenity(args)
class _AuthError(Exception):
def __init__(self, message="Authentication failed"):
super().__init__(message)
class PhysicalPasskey:
class InputDataError(Exception):
def __init__(self, message=""):
super().__init__(f"Input data insufficient or malformed: {message}")
AuthenticationError = _AuthError
def __init__(self):
from fido2.hid import CtapHidDevice
devices = list(CtapHidDevice.list_devices())
if not devices:
raise RuntimeError("No FIDO2 devices found")
self.device = devices[0]
def _get_client(self, origin):
from fido2.client import Fido2Client, DefaultClientDataCollector, UserInteraction
device = self.device
class ZenityInteraction(UserInteraction):
def prompt_up(self):
_zenity(["--notification", "--text", "Touch your security key..."], timeout=1)
def request_pin(self, permissions, rp_id):
pin = _zenity_entry(
"Physical WebAuthn",
f"Enter PIN for your security key\n\n{device}",
hide=True
)
if pin is None:
raise _AuthError("PIN prompt cancelled")
return pin
collector = DefaultClientDataCollector(origin)
return Fido2Client(self.device, collector, user_interaction=ZenityInteraction())
def create(self, create_options, origin=""):
from fido2.utils import websafe_encode, websafe_decode
options = create_options
if not origin:
origin = f'https://{options["rp"]["id"]}'
if not origin:
raise self.InputDataError("origin")
client = self._get_client(origin)
options["challenge"] = websafe_decode(options["challenge"])
options["user"]["id"] = websafe_decode(options["user"]["id"])
for cred in options.get("excludeCredentials", []):
cred["id"] = websafe_decode(cred["id"])
reg = client.make_credential(options)
return {
"authenticatorAttachment": "cross-platform",
"id": reg.id,
"rawId": reg.id,
"type": "public-key",
"response": {
"attestationObject": _b64url_encode(bytes(reg.response.attestation_object)),
"clientDataJSON": _b64url_encode(bytes(reg.response.client_data)),
},
}
def get(self, get_options, origin=""):
from fido2.utils import websafe_encode, websafe_decode
options = get_options
if not origin:
origin = f'https://{options["rpId"]}'
if not origin:
raise self.InputDataError("origin")
client = self._get_client(origin)
options["challenge"] = websafe_decode(options["challenge"])
for cred in options.get("allowCredentials", []):
cred["id"] = websafe_decode(cred["id"])
assertion = client.get_assertion(options).get_response(0)
return {
"authenticatorAttachment": "cross-platform",
"id": assertion.id,
"rawId": assertion.id,
"type": "public-key",
"response": {
"authenticatorData": _b64url_encode(bytes(assertion.response.authenticator_data)),
"clientDataJSON": _b64url_encode(bytes(assertion.response.client_data)),
"signature": _b64url_encode(bytes(assertion.response.signature)),
"userHandle": _b64url_encode(bytes(assertion.response.user_handle)) if assertion.response.user_handle else None,
},
}
class VirtualPasskey:
SCRYPT_N = 2**18
SCRYPT_R = 8
SCRYPT_P = 1
SCRYPT_KEYLEN = 32
def __init__(self, file: str = "passkey.json"):
self.file = file
self.credentials = {}
class InputDataError(Exception):
def __init__(self, message=""):
super().__init__(f"Input data insufficient or malformed: {message}")
class CredNotFoundError(Exception):
def __init__(self, message="No matching credential found"):
super().__init__(message)
AuthenticationError = _AuthError
def _ask_password(self, title: str = "Virtual WebAuthn") -> str:
if not os.path.exists(self.file):
log.info("No credential file, prompting new password")
pw = _zenity_password(f"{title} — Set Password")
pw2 = _zenity_password(f"{title} — Confirm Password")
if pw != pw2:
raise self.AuthenticationError("Passwords do not match")
self._save_credentials(pw)
log.info("Created credential file %s", self.file)
return pw
log.debug("Prompting password for %s", self.file)
return _zenity_password(title)
def _derive_key(self, password: str, salt: bytes) -> bytes:
return hashlib.scrypt(
password.encode(), salt=salt,
n=self.SCRYPT_N, r=self.SCRYPT_R, p=self.SCRYPT_P, dklen=self.SCRYPT_KEYLEN,
maxmem=128 * self.SCRYPT_N * self.SCRYPT_R * 2,
)
def _load_credentials(self, password: str) -> dict:
if not os.path.exists(self.file):
log.debug("Credential file not found, starting fresh")
return {}
with open(self.file, 'r') as f:
try:
envelope = json.load(f)
except (json.JSONDecodeError, ValueError):
log.warning("Credential file is corrupted, starting fresh")
return {}
# Unencrypted legacy format
if "salt" not in envelope:
log.debug("Loaded unencrypted legacy credentials")
return envelope
log.debug("Deriving key and decrypting credentials")
salt = _b64url_decode(envelope["salt"])
nonce = _b64url_decode(envelope["nonce"])
ciphertext = _b64url_decode(envelope["ciphertext"])
tag = _b64url_decode(envelope["tag"])
key = self._derive_key(password, salt)
cipher = AES.new(key, AES.MODE_GCM, nonce=nonce)
try:
plaintext = cipher.decrypt_and_verify(ciphertext, tag)
except (ValueError, KeyError):
raise self.AuthenticationError("Wrong password")
creds = json.loads(plaintext.decode())
log.debug("Decrypted %d credentials", len(creds))
return creds
def _save_credentials(self, password: str):
log.debug("Encrypting and saving %d credentials to %s", len(self.credentials), self.file)
salt = os.urandom(32)
nonce = os.urandom(12)
key = self._derive_key(password, salt)
cipher = AES.new(key, AES.MODE_GCM, nonce=nonce)
plaintext = json.dumps(self.credentials, indent=4).encode()
ciphertext, tag = cipher.encrypt_and_digest(plaintext)
envelope = {
"salt": _b64url_encode(salt),
"nonce": _b64url_encode(nonce),
"ciphertext": _b64url_encode(ciphertext),
"tag": _b64url_encode(tag),
}
with open(self.file, 'w') as f:
json.dump(envelope, f, indent=4)
log.debug("Credentials saved")
@staticmethod
def _build_authenticator_data(
rp_id: bytes, counter: int = 0,
user_present: bool = True,
user_verified: bool = True,
credential_data: Optional[bytes] = None,
) -> bytes:
rp_id_hash = hashlib.sha256(rp_id).digest()
flags = 0
if user_present:
flags |= 0x01
if user_verified:
flags |= 0x04
if credential_data is not None:
flags |= 0x40
auth_data = rp_id_hash + bytes([flags]) + struct.pack(">I", counter)
if credential_data is not None:
auth_data += credential_data
return auth_data
@staticmethod
def _cose_public_key(key) -> bytes:
x = key.pointQ.x.to_bytes(32, byteorder='big')
y = key.pointQ.y.to_bytes(32, byteorder='big')
return cbor2.dumps({1: 2, 3: -7, -1: 1, -2: x, -3: y})
def _find_credential(self, data: Dict[str, Any]) -> tuple:
allowed = data.get("allowCredentials") or []
if allowed:
for entry in allowed:
cred_id = entry["id"]
if cred_id in self.credentials:
return cred_id, self.credentials[cred_id]
raise self.CredNotFoundError()
rp_id = data.get("rpId", "")
for cred_id, cred_data in self.credentials.items():
stored_rp = _b64url_decode(cred_data["rp_id"]).decode('utf-8', errors='ignore')
if stored_rp == rp_id:
return cred_id, cred_data
raise self.CredNotFoundError()
def create(self, data: Dict[str, Any], origin: str = "") -> Dict[str, Any]:
password = self._ask_password("Virtual WebAuthn — Create Credential")
self.credentials = self._load_credentials(password)
challenge = data.get("challenge")
if isinstance(challenge, str):
challenge = _b64url_decode(challenge)
rp = data.get("rp", {})
user = data.get("user", {})
alg = -7
for param in data.get("pubKeyCredParams", []):
if param.get("type") == "public-key" and param.get("alg") == -7:
break
if not origin:
origin = data.get("origin")
if not origin:
raise self.InputDataError("origin")
rp_id = rp.get("id", "").encode()
user_id = user.get("id")
if isinstance(user_id, str):
user_id = _b64url_decode(user_id)
key = ECC.generate(curve='P-256')
credential_id = os.urandom(16)
credential_id_b64 = _b64url_encode(credential_id)
cose_pubkey = self._cose_public_key(key)
attested_data = (
b'\x00' * 16
+ struct.pack(">H", len(credential_id))
+ credential_id
+ cose_pubkey
)
auth_data = self._build_authenticator_data(rp_id, counter=0, credential_data=attested_data)
attestation_cbor = cbor2.dumps({
"fmt": "none",
"authData": auth_data,
"attStmt": {}
})
client_data_json = json.dumps({
"challenge": _b64url_encode(challenge),
"origin": origin,
"type": "webauthn.create",
"crossOrigin": False,
}).encode()
self.credentials[credential_id_b64] = {
"private_key": key.export_key(format='PEM'),
"rp_id": _b64url_encode(rp_id),
"user_id": _b64url_encode(user_id),
"user_name": user.get('displayName', ''),
"created": int(time.time()),
"counter": 0,
}
self._save_credentials(password)
return {
"authenticatorAttachment": "cross-platform",
"id": credential_id_b64,
"rawId": credential_id_b64,
"type": "public-key",
"response": {
"attestationObject": _b64url_encode(attestation_cbor),
"clientDataJSON": _b64url_encode(client_data_json),
"authenticatorData": _b64url_encode(auth_data),
"publicKey": _b64url_encode(cose_pubkey),
"pubKeyAlgo": str(alg),
"transports": ["internal"],
},
}
def get(self, data: Dict[str, Any], origin: str = "") -> Dict[str, Any]:
password = self._ask_password("Virtual WebAuthn — Authenticate")
self.credentials = self._load_credentials(password)
challenge = data.get("challenge")
if isinstance(challenge, str):
challenge = _b64url_decode(challenge)
credential_id_b64, cred = self._find_credential(data)
rp_id = data.get("rpId", "").encode('utf-8')
if not rp_id:
raise self.InputDataError("rpId")
if not origin:
origin = data.get("origin")
if not origin:
raise self.InputDataError("origin")
counter = cred.get("counter", 0) + 1
cred["counter"] = counter
auth_data = self._build_authenticator_data(rp_id, counter=counter)
client_data = json.dumps({
"type": "webauthn.get",
"challenge": _b64url_encode(challenge),
"origin": origin,
"crossOrigin": False,
}, separators=(',', ':')).encode()
client_data_hash = hashlib.sha256(client_data).digest()
key = ECC.import_key(cred["private_key"])
h = SHA256.new(auth_data + client_data_hash)
signature = DSS.new(key, 'fips-186-3', encoding='der').sign(h)
self._save_credentials(password)
return {
"authenticatorAttachment": "cross-platform",
"id": credential_id_b64,
"rawId": credential_id_b64,
"type": "public-key",
"response": {
"authenticatorData": _b64url_encode(auth_data),
"clientDataJSON": _b64url_encode(client_data),
"signature": _b64url_encode(signature),
},
}
Passkey = VirtualPasskey
if __name__ == "__main__":
import requests
sess = requests.Session()
passkey = Passkey()
reg_payload = {
"algorithms": ["es256"], "attachment": "all", "attestation": "none",
"discoverable_credential": "preferred", "hints": [],
"user_verification": "preferred", "username": "test",
}
options = sess.post("https://webauthn.io/registration/options", json=reg_payload).json()
cred = passkey.create(options, origin="https://webauthn.io")
cred["rawId"] = cred["id"]
result = sess.post("https://webauthn.io/registration/verification",
json={"response": cred, "username": "test"}).json()
print("Registration:", result)
sess.get("https://webauthn.io/logout")
auth_payload = {"username": "test", "user_verification": "preferred", "hints": []}
options = sess.post("https://webauthn.io/authentication/options", json=auth_payload).json()
assertion = passkey.get(options, origin="https://webauthn.io")
assertion["rawId"] = assertion["id"]
result = sess.post("https://webauthn.io/authentication/verification",
json={"response": assertion, "username": "test"}).json()
print("Authentication:", result)

704
src/main.rs Normal file
View File

@@ -0,0 +1,704 @@
use aes_gcm::{aead::Aead, Aes256Gcm, KeyInit, Nonce};
use base64ct::{Base64UrlUnpadded, Encoding};
use ecdsa::signature::Signer;
use log::{info, warn};
use p256::ecdsa::{DerSignature, SigningKey};
use p256::pkcs8::{DecodePrivateKey, EncodePrivateKey};
use rand::RngCore;
use sha2::{Digest, Sha256};
use std::collections::HashMap;
use std::fs;
use std::io::{self, Read, Write};
use std::path::PathBuf;
use std::time::{SystemTime, UNIX_EPOCH};
// --- Base64url ---
fn b64url_encode(data: &[u8]) -> String {
Base64UrlUnpadded::encode_string(data)
}
fn b64url_decode(s: &str) -> Result<Vec<u8>, String> {
Base64UrlUnpadded::decode_vec(s).map_err(|e| format!("base64 decode: {e}"))
}
// --- AES-GCM helpers ---
#[derive(serde::Serialize, serde::Deserialize, Clone)]
struct Encrypted {
nonce: String,
ciphertext: String,
}
fn aes_encrypt(key: &[u8; 32], plaintext: &[u8]) -> Encrypted {
let cipher = Aes256Gcm::new_from_slice(key).unwrap();
let mut nonce_bytes = [0u8; 12];
rand::thread_rng().fill_bytes(&mut nonce_bytes);
let nonce = Nonce::from_slice(&nonce_bytes);
let ct = cipher.encrypt(nonce, plaintext).expect("encrypt");
Encrypted {
nonce: b64url_encode(&nonce_bytes),
ciphertext: b64url_encode(&ct), // includes tag
}
}
fn aes_decrypt(key: &[u8; 32], enc: &Encrypted) -> Result<Vec<u8>, String> {
let cipher = Aes256Gcm::new_from_slice(key).unwrap();
let nonce_bytes = b64url_decode(&enc.nonce)?;
let ct = b64url_decode(&enc.ciphertext)?;
let nonce = Nonce::from_slice(&nonce_bytes);
cipher
.decrypt(nonce, ct.as_ref())
.map_err(|_| "Decryption failed".to_string())
}
// --- Key derivation ---
fn derive_key(password: &str, salt: &[u8]) -> [u8; 32] {
let params = scrypt::Params::new(18, 8, 1, 32).expect("scrypt params");
let mut key = [0u8; 32];
scrypt::scrypt(password.as_bytes(), salt, &params, &mut key).expect("scrypt");
key
}
fn random_bytes<const N: usize>() -> [u8; N] {
let mut buf = [0u8; N];
rand::thread_rng().fill_bytes(&mut buf);
buf
}
// --- CBOR ---
fn cbor_encode(value: &ciborium::Value) -> Vec<u8> {
let mut buf = Vec::new();
ciborium::into_writer(value, &mut buf).expect("cbor encode");
buf
}
fn cose_public_key(key: &SigningKey) -> Vec<u8> {
let point = key.verifying_key().to_encoded_point(false);
let x = point.x().unwrap();
let y = point.y().unwrap();
use ciborium::Value as V;
cbor_encode(&V::Map(vec![
(V::Integer(1.into()), V::Integer(2.into())),
(V::Integer(3.into()), V::Integer((-7).into())),
(V::Integer((-1).into()), V::Integer(1.into())),
(V::Integer((-2).into()), V::Bytes(x.to_vec())),
(V::Integer((-3).into()), V::Bytes(y.to_vec())),
]))
}
// --- Authenticator data ---
fn build_auth_data(rp_id: &[u8], counter: u32, credential_data: Option<&[u8]>) -> Vec<u8> {
let rp_id_hash = Sha256::digest(rp_id);
let mut flags: u8 = 0x01 | 0x04; // UP + UV
if credential_data.is_some() {
flags |= 0x40;
}
let mut data = Vec::new();
data.extend_from_slice(&rp_id_hash);
data.push(flags);
data.extend_from_slice(&counter.to_be_bytes());
if let Some(cd) = credential_data {
data.extend_from_slice(cd);
}
data
}
// --- Credential storage ---
/// Plaintext metadata for discovery
#[derive(serde::Serialize, serde::Deserialize, Clone)]
struct CredentialMeta {
rp_id: String,
user_name: String,
created: u64,
}
/// Secret data encrypted by master key
#[derive(serde::Serialize, serde::Deserialize)]
struct CredentialSecret {
private_key_pem: String,
user_id: String,
counter: u32,
}
/// Per-credential entry in the store
#[derive(serde::Serialize, serde::Deserialize, Clone)]
struct StoredCredential {
#[serde(flatten)]
meta: CredentialMeta,
encrypted: Encrypted,
}
/// Wrapped master key (encrypted by user password)
#[derive(serde::Serialize, serde::Deserialize, Clone)]
struct WrappedMasterKey {
salt: String, // scrypt salt for password -> wrapping key
#[serde(flatten)]
encrypted: Encrypted,
}
/// The credential file format
#[derive(serde::Serialize, serde::Deserialize)]
struct CredentialStore {
master_key: WrappedMasterKey,
credentials: HashMap<String, StoredCredential>,
}
/// Session file format
#[derive(serde::Serialize, serde::Deserialize)]
struct SessionFile {
session_id: String,
wrapped_master_key: Encrypted,
expires: u64,
}
fn now_secs() -> u64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap()
.as_secs()
}
// --- Master key management ---
fn create_store(password: &str) -> Result<(CredentialStore, [u8; 32]), String> {
let master_key: [u8; 32] = random_bytes();
let salt: [u8; 32] = random_bytes();
let wrapping_key = derive_key(password, &salt);
let encrypted = aes_encrypt(&wrapping_key, &master_key);
let store = CredentialStore {
master_key: WrappedMasterKey {
salt: b64url_encode(&salt),
encrypted,
},
credentials: HashMap::new(),
};
Ok((store, master_key))
}
fn unwrap_master_key(store: &CredentialStore, password: &str) -> Result<[u8; 32], String> {
let salt = b64url_decode(&store.master_key.salt)?;
let wrapping_key = derive_key(password, &salt);
let plain = aes_decrypt(&wrapping_key, &store.master_key.encrypted)
.map_err(|_| "Wrong password".to_string())?;
let mut key = [0u8; 32];
if plain.len() != 32 {
return Err("Corrupted master key".into());
}
key.copy_from_slice(&plain);
Ok(key)
}
fn encrypt_credential(master_key: &[u8; 32], secret: &CredentialSecret) -> Encrypted {
let plain = serde_json::to_vec(secret).unwrap();
aes_encrypt(master_key, &plain)
}
fn decrypt_credential(
master_key: &[u8; 32],
cred: &StoredCredential,
) -> Result<CredentialSecret, String> {
let plain = aes_decrypt(master_key, &cred.encrypted)?;
serde_json::from_slice(&plain).map_err(|e| format!("credential parse: {e}"))
}
// --- File I/O ---
fn load_store(path: &PathBuf) -> Result<CredentialStore, String> {
let data = fs::read_to_string(path).map_err(|e| format!("read: {e}"))?;
serde_json::from_str(&data).map_err(|e| format!("parse: {e}"))
}
fn save_store(path: &PathBuf, store: &CredentialStore) -> Result<(), String> {
if let Some(parent) = path.parent() {
fs::create_dir_all(parent).map_err(|e| format!("mkdir: {e}"))?;
}
let data = serde_json::to_string_pretty(store).unwrap();
fs::write(path, data).map_err(|e| format!("write: {e}"))?;
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
let _ = fs::set_permissions(path, fs::Permissions::from_mode(0o600));
}
Ok(())
}
// --- Session management ---
const SESSION_TTL: u64 = 300; // 5 minutes
fn create_session(
master_key: &[u8; 32],
session_dir: &PathBuf,
) -> Result<String, String> {
let session_key: [u8; 32] = random_bytes();
let session_id = b64url_encode(&random_bytes::<16>());
let wrapped = aes_encrypt(&session_key, master_key);
let session = SessionFile {
session_id: session_id.clone(),
wrapped_master_key: wrapped,
expires: now_secs() + SESSION_TTL,
};
let session_path = session_dir.join("session.json");
let data = serde_json::to_string_pretty(&session).unwrap();
fs::write(&session_path, data).map_err(|e| format!("write session: {e}"))?;
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
let _ = fs::set_permissions(&session_path, fs::Permissions::from_mode(0o600));
}
// Return session_key as b64url for the extension to hold
Ok(b64url_encode(&session_key))
}
fn resume_session(
session_key_b64: &str,
session_dir: &PathBuf,
) -> Result<[u8; 32], String> {
let session_path = session_dir.join("session.json");
if !session_path.exists() {
return Err("No active session".into());
}
let data = fs::read_to_string(&session_path).map_err(|e| format!("read session: {e}"))?;
let session: SessionFile =
serde_json::from_str(&data).map_err(|e| format!("parse session: {e}"))?;
if now_secs() > session.expires {
let _ = fs::remove_file(&session_path);
return Err("Session expired".into());
}
let session_key_bytes = b64url_decode(session_key_b64)?;
let mut session_key = [0u8; 32];
if session_key_bytes.len() != 32 {
return Err("Invalid session key".into());
}
session_key.copy_from_slice(&session_key_bytes);
let master_key_bytes = aes_decrypt(&session_key, &session.wrapped_master_key)
.map_err(|_| "Invalid session key".to_string())?;
let mut master_key = [0u8; 32];
if master_key_bytes.len() != 32 {
return Err("Corrupted session".into());
}
master_key.copy_from_slice(&master_key_bytes);
Ok(master_key)
}
fn revoke_session(session_dir: &PathBuf) {
let _ = fs::remove_file(session_dir.join("session.json"));
}
// --- Resolve master key from password or session ---
fn get_master_key(
msg: &serde_json::Value,
cred_file: &PathBuf,
session_dir: &PathBuf,
) -> Result<([u8; 32], CredentialStore, Option<String>), String> {
let password = msg["password"].as_str().unwrap_or("");
let session_key = msg["sessionKey"].as_str().unwrap_or("");
// Try session first
if !session_key.is_empty() {
let master_key = resume_session(session_key, session_dir)?;
let store = if cred_file.exists() {
load_store(cred_file)?
} else {
return Err("No credential file".into());
};
return Ok((master_key, store, None));
}
// Otherwise use password
if password.is_empty() {
return Err("Password or session key required".into());
}
if !cred_file.exists() {
// First time setup
let (store, master_key) = create_store(password)?;
save_store(cred_file, &store)?;
info!("Created new credential store");
let new_session = create_session(&master_key, session_dir)?;
return Ok((master_key, store, Some(new_session)));
}
let store = load_store(cred_file)?;
let master_key = unwrap_master_key(&store, password)?;
let new_session = create_session(&master_key, session_dir)?;
Ok((master_key, store, Some(new_session)))
}
// --- WebAuthn create ---
fn webauthn_create(
data: &serde_json::Value,
origin: &str,
master_key: &[u8; 32],
store: &mut CredentialStore,
cred_file: &PathBuf,
) -> Result<serde_json::Value, String> {
let challenge = b64url_decode(data["challenge"].as_str().ok_or("missing challenge")?)?;
let rp = &data["rp"];
let user = &data["user"];
let rp_id_str = rp["id"].as_str().unwrap_or("");
let rp_id = rp_id_str.as_bytes();
let origin = if origin.is_empty() {
data["origin"].as_str().ok_or("missing origin")?.to_string()
} else {
origin.to_string()
};
let user_id_str = user["id"].as_str().unwrap_or("");
let user_id = b64url_decode(user_id_str)?;
let signing_key = SigningKey::random(&mut rand::thread_rng());
let credential_id: [u8; 16] = random_bytes();
let credential_id_b64 = b64url_encode(&credential_id);
let cose_pubkey = cose_public_key(&signing_key);
let mut attested_data = vec![0u8; 16];
attested_data.extend_from_slice(&(credential_id.len() as u16).to_be_bytes());
attested_data.extend_from_slice(&credential_id);
attested_data.extend_from_slice(&cose_pubkey);
let auth_data = build_auth_data(rp_id, 0, Some(&attested_data));
use ciborium::Value as V;
let attestation_cbor = cbor_encode(&V::Map(vec![
(V::Text("fmt".into()), V::Text("none".into())),
(V::Text("authData".into()), V::Bytes(auth_data.clone())),
(V::Text("attStmt".into()), V::Map(vec![])),
]));
let client_data_json = serde_json::json!({
"challenge": b64url_encode(&challenge),
"origin": origin,
"type": "webauthn.create",
"crossOrigin": false,
})
.to_string();
let pem = signing_key
.to_pkcs8_pem(p256::pkcs8::LineEnding::LF)
.map_err(|e| format!("pem: {e}"))?;
let secret = CredentialSecret {
private_key_pem: pem.to_string(),
user_id: b64url_encode(&user_id),
counter: 0,
};
store.credentials.insert(
credential_id_b64.clone(),
StoredCredential {
meta: CredentialMeta {
rp_id: rp_id_str.to_string(),
user_name: user["displayName"]
.as_str()
.or(user["name"].as_str())
.unwrap_or("")
.to_string(),
created: now_secs(),
},
encrypted: encrypt_credential(master_key, &secret),
},
);
save_store(cred_file, store)?;
Ok(serde_json::json!({
"authenticatorAttachment": "cross-platform",
"id": credential_id_b64,
"rawId": credential_id_b64,
"type": "public-key",
"response": {
"attestationObject": b64url_encode(&attestation_cbor),
"clientDataJSON": b64url_encode(client_data_json.as_bytes()),
"authenticatorData": b64url_encode(&auth_data),
"publicKey": b64url_encode(&cose_pubkey),
"pubKeyAlgo": "-7",
"transports": ["internal"],
},
}))
}
// --- WebAuthn get ---
fn webauthn_get(
data: &serde_json::Value,
origin: &str,
master_key: &[u8; 32],
store: &mut CredentialStore,
cred_file: &PathBuf,
) -> Result<serde_json::Value, String> {
let challenge = b64url_decode(data["challenge"].as_str().ok_or("missing challenge")?)?;
let rp_id_str = data["rpId"].as_str().unwrap_or("");
if rp_id_str.is_empty() {
return Err("missing rpId".into());
}
let rp_id = rp_id_str.as_bytes();
let origin = if origin.is_empty() {
data["origin"].as_str().ok_or("missing origin")?.to_string()
} else {
origin.to_string()
};
// Find credential by allowCredentials or rpId discovery
let allowed = data["allowCredentials"].as_array();
let cred_id_b64 = if let Some(allowed) = allowed {
let mut found = None;
for entry in allowed {
if let Some(id) = entry["id"].as_str() {
if store.credentials.contains_key(id) {
found = Some(id.to_string());
break;
}
}
}
found.ok_or("No matching credential found")?
} else {
let mut found = None;
for (cid, c) in &store.credentials {
if c.meta.rp_id == rp_id_str {
found = Some(cid.clone());
break;
}
}
found.ok_or("No matching credential found")?
};
let cred = store
.credentials
.get(&cred_id_b64)
.ok_or("Credential not found")?;
let mut secret = decrypt_credential(master_key, cred)?;
secret.counter += 1;
let auth_data = build_auth_data(rp_id, secret.counter, None);
let client_data = serde_json::json!({
"type": "webauthn.get",
"challenge": b64url_encode(&challenge),
"origin": origin,
"crossOrigin": false,
});
let client_data_bytes = client_data.to_string().into_bytes();
let client_data_hash = Sha256::digest(&client_data_bytes);
let signing_key = SigningKey::from_pkcs8_pem(&secret.private_key_pem)
.map_err(|e| format!("key: {e}"))?;
let mut signed_data = auth_data.clone();
signed_data.extend_from_slice(&client_data_hash);
let signature: DerSignature = signing_key.sign(&signed_data);
// Re-encrypt with updated counter
let updated = StoredCredential {
meta: cred.meta.clone(),
encrypted: encrypt_credential(master_key, &secret),
};
store.credentials.insert(cred_id_b64.clone(), updated);
save_store(cred_file, store)?;
Ok(serde_json::json!({
"authenticatorAttachment": "cross-platform",
"id": cred_id_b64,
"rawId": cred_id_b64,
"type": "public-key",
"response": {
"authenticatorData": b64url_encode(&auth_data),
"clientDataJSON": b64url_encode(&client_data_bytes),
"signature": b64url_encode(&signature.to_bytes()),
"userHandle": secret.user_id,
},
}))
}
// --- List credentials (no password needed) ---
fn list_credentials(cred_file: &PathBuf, rp_id: &str) -> Result<serde_json::Value, String> {
if !cred_file.exists() {
return Ok(serde_json::json!([]));
}
let store = load_store(cred_file)?;
let matches: Vec<_> = store
.credentials
.iter()
.filter(|(_, c)| rp_id.is_empty() || c.meta.rp_id == rp_id)
.map(|(id, c)| {
serde_json::json!({
"id": id,
"rp_id": c.meta.rp_id,
"user_name": c.meta.user_name,
"created": c.meta.created,
})
})
.collect();
Ok(serde_json::json!(matches))
}
// --- Native Messaging ---
fn read_message() -> Option<serde_json::Value> {
let mut len_buf = [0u8; 4];
if io::stdin().read_exact(&mut len_buf).is_err() {
return None;
}
let len = u32::from_le_bytes(len_buf) as usize;
let mut buf = vec![0u8; len];
if io::stdin().read_exact(&mut buf).is_err() {
return None;
}
serde_json::from_slice(&buf).ok()
}
fn send_message(msg: &serde_json::Value) {
let data = serde_json::to_vec(msg).unwrap();
let len = (data.len() as u32).to_le_bytes();
let stdout = io::stdout();
let mut out = stdout.lock();
let _ = out.write_all(&len);
let _ = out.write_all(&data);
let _ = out.flush();
}
fn handle(msg: &serde_json::Value, cred_file: &PathBuf, session_dir: &PathBuf) -> serde_json::Value {
let msg_type = msg["type"].as_str().unwrap_or("");
let msg_id = &msg["id"];
// No-auth operations
match msg_type {
"ping" => {
return serde_json::json!({
"id": msg_id,
"success": true,
"data": {"status": "ok"},
});
}
"status" => {
let exists = cred_file.exists();
return serde_json::json!({
"id": msg_id,
"success": true,
"data": {"needsSetup": !exists},
});
}
"list" => {
let rp_id = msg["rpId"].as_str().unwrap_or("");
match list_credentials(cred_file, rp_id) {
Ok(data) => {
return serde_json::json!({"id": msg_id, "success": true, "data": data});
}
Err(e) => {
return serde_json::json!({"id": msg_id, "success": false, "error": e});
}
}
}
"revoke" => {
revoke_session(session_dir);
return serde_json::json!({
"id": msg_id,
"success": true,
"data": {"revoked": true},
});
}
_ => {}
}
// Auth-required operations
let data = &msg["data"];
let options = &data["publicKey"];
let origin = data["origin"].as_str().unwrap_or("");
let (master_key, mut store, new_session) = match get_master_key(msg, cred_file, session_dir) {
Ok(v) => v,
Err(e) => {
warn!("auth error: {}", e);
return serde_json::json!({"id": msg_id, "success": false, "error": e});
}
};
let rp = options["rp"]["id"]
.as_str()
.or(options["rpId"].as_str())
.unwrap_or("");
info!("{} rp={} origin={}", msg_type, rp, origin);
let result = match msg_type {
"create" => webauthn_create(options, origin, &master_key, &mut store, cred_file),
"get" => webauthn_get(options, origin, &master_key, &mut store, cred_file),
_ => Err(format!("Unknown type: {msg_type}")),
};
match result {
Ok(data) => {
info!(" success");
let mut resp = serde_json::json!({"id": msg_id, "success": true, "data": data});
if let Some(sk) = new_session {
resp["sessionKey"] = serde_json::Value::String(sk);
}
resp
}
Err(e) => {
warn!(" error: {}", e);
serde_json::json!({"id": msg_id, "success": false, "error": e})
}
}
}
fn main() {
let cred_dir = dirs::home_dir()
.unwrap_or_else(|| PathBuf::from("."))
.join(".passkeys");
let cred_file = std::env::var("VWEBAUTHN_CRED_FILE")
.map(PathBuf::from)
.unwrap_or_else(|_| cred_dir.join("credentials.json"));
let _ = fs::create_dir_all(&cred_dir);
let log_file = cred_dir.join("host.log");
if let Ok(file) = fs::OpenOptions::new()
.create(true)
.append(true)
.open(&log_file)
{
env_logger::Builder::new()
.filter_level(
if std::env::var("VWEBAUTHN_VERBOSE").unwrap_or_default() == "1" {
log::LevelFilter::Debug
} else {
log::LevelFilter::Info
},
)
.target(env_logger::Target::Pipe(Box::new(file)))
.format_timestamp_secs()
.init();
}
info!("Host started, cred_file={}", cred_file.display());
loop {
let msg = match read_message() {
Some(m) => m,
None => break,
};
let response = handle(&msg, &cred_file, &cred_dir);
send_message(&response);
}
info!("Host exiting");
}

9
virtual_webauthn.json Normal file
View File

@@ -0,0 +1,9 @@
{
"name": "com.example.virtual_webauthn",
"description": "Virtual WebAuthn Native Messaging Host",
"path": "/PLACEHOLDER/virtual-webauthn",
"type": "stdio",
"allowed_extensions": [
"virtual-webauthn@local"
]
}