v0.0.48 - Added OTP one-time pad encryption (otp_encrypt/otp_decrypt verbs), HTTP listener mode (--listen http:HOST:PORT), interactive OTP pad auto-scan on USB drives, raised SERVER_MAX_MSG_SIZE to 16MB, updated README with curl examples and current API documentation
This commit is contained in:
Laan Tungir
213
tools/otp_roundtrip_test.py
Normal file
213
tools/otp_roundtrip_test.py
Normal file
@@ -0,0 +1,213 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
otp_roundtrip_test.py — end-to-end test of otp_encrypt / otp_decrypt verbs.
|
||||
|
||||
Sends framed JSON-RPC requests to nsigner --listen stdio and checks the
|
||||
round-trip: plaintext -> otp_encrypt -> otp_decrypt -> recovered plaintext.
|
||||
|
||||
Framing: 4-byte big-endian length prefix + JSON payload.
|
||||
|
||||
Usage: python3 tools/otp_roundtrip_test.py
|
||||
"""
|
||||
import base64
|
||||
import json
|
||||
import os
|
||||
import struct
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import time
|
||||
|
||||
NSIGNER = "./build/nsigner"
|
||||
PAD_DIR = "/media/user/Music/pads"
|
||||
PAD_SPEC = "333e9902db839d9d"
|
||||
MNEMONIC_FILE = ".test_mnemonic"
|
||||
|
||||
|
||||
def send_framed(proc, obj):
|
||||
payload = json.dumps(obj).encode()
|
||||
proc.stdin.write(struct.pack(">I", len(payload)))
|
||||
proc.stdin.write(payload)
|
||||
proc.stdin.flush()
|
||||
|
||||
|
||||
def recv_framed(proc):
|
||||
"""Read a framed response, skipping any banner text the signer writes
|
||||
to stdout before the first frame. The banner is line-based ASCII; a
|
||||
valid frame starts with a 4-byte big-endian length followed by '{'."""
|
||||
# Read 4 bytes at a time, sliding window, until we find a frame header.
|
||||
buf = b""
|
||||
while True:
|
||||
b = proc.stdout.read(1)
|
||||
if not b:
|
||||
return None
|
||||
buf = (buf + b)[-4:]
|
||||
if len(buf) < 4:
|
||||
continue
|
||||
(length,) = struct.unpack(">I", buf)
|
||||
# Sanity: frame length should be reasonable (1..1MB) and the next
|
||||
# byte after the header should be '{' (start of JSON).
|
||||
if 1 <= length <= 1024 * 1024:
|
||||
# Peek: read one more byte to check for '{'.
|
||||
peek = proc.stdout.read(1)
|
||||
if peek == b'{':
|
||||
body = peek + proc.stdout.read(length - 1)
|
||||
return json.loads(body.decode())
|
||||
else:
|
||||
# Not a frame; prepend peek to the stream by including it
|
||||
# in the sliding window.
|
||||
buf = (buf + peek)[-4:]
|
||||
# Otherwise keep scanning.
|
||||
|
||||
|
||||
def main():
|
||||
if not os.path.isfile(NSIGNER):
|
||||
print(f"ERROR: {NSIGNER} not found. Run 'make dev' first.")
|
||||
return 1
|
||||
|
||||
with open(MNEMONIC_FILE) as f:
|
||||
mnemonic = f.read().strip()
|
||||
|
||||
plaintext = b"Hello, OTP world!"
|
||||
pt_b64 = base64.b64encode(plaintext).decode()
|
||||
print(f"Plaintext: {plaintext.decode()}")
|
||||
print(f"Plaintext base64: {pt_b64}")
|
||||
|
||||
# Write the mnemonic to a fixed temp file and pass it as fd 3 to nsigner
|
||||
# via a bash wrapper (so stdin stays free for framed requests).
|
||||
mnem_path = ".test_mnemonic_otp_roundtrip.tmp"
|
||||
with open(mnem_path, "w") as f:
|
||||
f.write(mnemonic + "\n")
|
||||
|
||||
def run_one_request(req_obj):
|
||||
"""Run nsigner in stdio mode for a single framed request/response.
|
||||
Returns the parsed JSON response or None."""
|
||||
shell_cmd = (
|
||||
f"exec 3<{mnem_path}; "
|
||||
f"exec {NSIGNER} --listen stdio --mnemonic-fd 3 "
|
||||
f"--otp-pad-dir {PAD_DIR} --otp-pad {PAD_SPEC} "
|
||||
f"--otp-allow-blkback --allow-all"
|
||||
)
|
||||
proc = subprocess.Popen(
|
||||
["bash", "-c", shell_cmd],
|
||||
stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
|
||||
text=False,
|
||||
)
|
||||
time.sleep(1.5) # let the signer bind the pad and be ready
|
||||
if proc.poll() is not None:
|
||||
err = proc.stderr.read().decode()
|
||||
print(f"ERROR: nsigner exited early (code {proc.returncode})")
|
||||
print(f"stderr: {err}")
|
||||
return None
|
||||
send_framed(proc, req_obj)
|
||||
resp = recv_framed(proc)
|
||||
proc.stdin.close()
|
||||
try:
|
||||
proc.wait(timeout=5)
|
||||
except subprocess.TimeoutExpired:
|
||||
proc.kill()
|
||||
return resp
|
||||
|
||||
# --- otp_encrypt ---
|
||||
print("\n=== Sending otp_encrypt ===")
|
||||
resp = run_one_request({
|
||||
"id": "1",
|
||||
"method": "otp_encrypt",
|
||||
"params": [pt_b64, {"encoding": "ascii"}],
|
||||
})
|
||||
print(f"Encrypt response: {json.dumps(resp)}")
|
||||
|
||||
if resp is None or "result" not in resp:
|
||||
print("ERROR: no result in encrypt response")
|
||||
try:
|
||||
os.unlink(mnem_path)
|
||||
except OSError:
|
||||
pass
|
||||
return 1
|
||||
|
||||
result_obj = json.loads(resp["result"])
|
||||
ciphertext = result_obj["ciphertext"]
|
||||
off_before = result_obj["pad_offset_before"]
|
||||
off_after = result_obj["pad_offset_after"]
|
||||
print(f"Pad offset: {off_before} -> {off_after} "
|
||||
f"(consumed {off_after - off_before} bytes)")
|
||||
print(f"Ciphertext (first 80 chars): {ciphertext[:80]}...")
|
||||
|
||||
# --- otp_decrypt (separate invocation; offset persists in .state) ---
|
||||
print("\n=== Sending otp_decrypt ===")
|
||||
resp2 = run_one_request({
|
||||
"id": "2",
|
||||
"method": "otp_decrypt",
|
||||
"params": [ciphertext, {"encoding": "ascii"}],
|
||||
})
|
||||
print(f"Decrypt response: {json.dumps(resp2)}")
|
||||
|
||||
if resp2 is None or "result" not in resp2:
|
||||
print("ERROR: no result in decrypt response")
|
||||
return 1
|
||||
|
||||
result2 = json.loads(resp2["result"])
|
||||
recovered_b64 = result2["plaintext"]
|
||||
recovered = base64.b64decode(recovered_b64)
|
||||
print(f"\nRecovered plaintext: {recovered.decode()}")
|
||||
|
||||
if recovered == plaintext:
|
||||
print("\n=== ASCII ROUND-TRIP SUCCESS ===")
|
||||
else:
|
||||
print("\n=== ASCII ROUND-TRIP FAILED ===")
|
||||
print(f"Expected: {plaintext.decode()}")
|
||||
print(f"Got: {recovered.decode()}")
|
||||
return 1
|
||||
|
||||
# --- Binary encoding round-trip ---
|
||||
# Reset the pad offset for a clean binary test.
|
||||
state_path = (f"{PAD_DIR}/{result_obj['pad_chksum']}.state")
|
||||
with open(state_path, "w") as sf:
|
||||
sf.write("offset=32\n")
|
||||
|
||||
print("\n=== Sending otp_encrypt (binary) ===")
|
||||
resp3 = run_one_request({
|
||||
"id": "3",
|
||||
"method": "otp_encrypt",
|
||||
"params": [pt_b64, {"encoding": "binary"}],
|
||||
})
|
||||
print(f"Binary encrypt response: {json.dumps(resp3)}")
|
||||
if resp3 is None or "result" not in resp3:
|
||||
print("ERROR: no result in binary encrypt response")
|
||||
return 1
|
||||
result3 = json.loads(resp3["result"])
|
||||
bin_b64 = result3["ciphertext"]
|
||||
# The binary ciphertext is base64-encoded in the JSON result.
|
||||
bin_blob = base64.b64decode(bin_b64)
|
||||
print(f"Binary blob size: {len(bin_blob)} bytes "
|
||||
f"(header 58 + padded data {len(bin_blob) - 58})")
|
||||
if not bin_blob[:4] == b"OTP\0":
|
||||
print("ERROR: binary blob missing OTP magic")
|
||||
return 1
|
||||
|
||||
print("\n=== Sending otp_decrypt (binary) ===")
|
||||
resp4 = run_one_request({
|
||||
"id": "4",
|
||||
"method": "otp_decrypt",
|
||||
"params": [bin_b64, {"encoding": "binary"}],
|
||||
})
|
||||
print(f"Binary decrypt response: {json.dumps(resp4)}")
|
||||
if resp4 is None or "result" not in resp4:
|
||||
print("ERROR: no result in binary decrypt response")
|
||||
return 1
|
||||
result4 = json.loads(resp4["result"])
|
||||
recovered2 = base64.b64decode(result4["plaintext"])
|
||||
print(f"\nRecovered plaintext (binary path): {recovered2.decode()}")
|
||||
|
||||
if recovered2 == plaintext:
|
||||
print("\n=== BINARY ROUND-TRIP SUCCESS ===")
|
||||
return 0
|
||||
else:
|
||||
print("\n=== BINARY ROUND-TRIP FAILED ===")
|
||||
print(f"Expected: {plaintext.decode()}")
|
||||
print(f"Got: {recovered2.decode()}")
|
||||
return 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user