--pin-source stdin prevents PIN visibility in `ps` output. Use tempfile.mkstemp for all temp files (unpredictable names, 0600 perms). Clean up temp files in finally block. Add tests for PIN not in args and mkstemp usage.
This commit is contained in:
parent
279a28515f
commit
c759597ad0
@ -1,12 +1,11 @@
|
|||||||
|
|
||||||
import subprocess, datetime, os, hashlib, ipaddress, re
|
import subprocess, datetime, os, hashlib, ipaddress, re, tempfile
|
||||||
from cryptography import x509
|
from cryptography import x509
|
||||||
from cryptography.hazmat.primitives import hashes, serialization
|
from cryptography.hazmat.primitives import hashes, serialization
|
||||||
from cryptography.hazmat.primitives.asymmetric import ec
|
from cryptography.hazmat.primitives.asymmetric import ec
|
||||||
from cryptography.x509.oid import NameOID
|
from cryptography.x509.oid import NameOID
|
||||||
from config import *
|
from config import *
|
||||||
|
|
||||||
TMP_DIR = "/var/lib/certauth/tmp"
|
|
||||||
os.makedirs(TMP_DIR, exist_ok=True)
|
os.makedirs(TMP_DIR, exist_ok=True)
|
||||||
|
|
||||||
def get_root_pub_key():
|
def get_root_pub_key():
|
||||||
@ -26,24 +25,40 @@ def der_len(n):
|
|||||||
elif n < 0x100: return bytes([0x81, n])
|
elif n < 0x100: return bytes([0x81, n])
|
||||||
return bytes([0x82, n>>8, n&0xff])
|
return bytes([0x82, n>>8, n&0xff])
|
||||||
|
|
||||||
|
def _make_temp_file(prefix: str, data: bytes = None):
|
||||||
|
"""Create temp file with unpredictable name in TMP_DIR."""
|
||||||
|
fd, path = tempfile.mkstemp(prefix=prefix, dir=TMP_DIR)
|
||||||
|
try:
|
||||||
|
if data is not None:
|
||||||
|
os.write(fd, data)
|
||||||
|
finally:
|
||||||
|
os.close(fd)
|
||||||
|
os.chmod(path, 0o600)
|
||||||
|
return path
|
||||||
|
|
||||||
|
|
||||||
def sign_tbs_with_yk(tbs_bytes, yk_pin, token_label="certauth Intermediate CA"):
|
def sign_tbs_with_yk(tbs_bytes, yk_pin, token_label="certauth Intermediate CA"):
|
||||||
tbs_file = os.path.join(TMP_DIR, "tbs_sign.der")
|
tbs_file = _make_temp_file("tbs_", tbs_bytes)
|
||||||
sig_file = os.path.join(TMP_DIR, "sig_out.bin")
|
sig_file = _make_temp_file("sig_")
|
||||||
with open(tbs_file, "wb") as f:
|
os.unlink(sig_file)
|
||||||
f.write(tbs_bytes)
|
try:
|
||||||
r = subprocess.run([
|
r = subprocess.run([
|
||||||
"sudo", "pkcs11-tool", "--module", PKCS11_MODULE,
|
"sudo", "pkcs11-tool", "--module", PKCS11_MODULE,
|
||||||
"--login", "--pin", yk_pin,
|
"--login", "--pin-source", "stdin",
|
||||||
"--sign", "--mechanism", "ECDSA-SHA384",
|
"--sign", "--mechanism", "ECDSA-SHA384",
|
||||||
"--token-label", token_label,
|
"--token-label", token_label,
|
||||||
"--label", "SIGN key",
|
"--label", "SIGN key",
|
||||||
"--input-file", tbs_file,
|
"--input-file", tbs_file,
|
||||||
"--output-file", sig_file
|
"--output-file", sig_file
|
||||||
], capture_output=True, text=True)
|
], input=yk_pin, capture_output=True, text=True)
|
||||||
if r.returncode != 0:
|
if r.returncode != 0:
|
||||||
return None, r.stderr
|
return None, r.stderr
|
||||||
with open(sig_file, "rb") as f:
|
with open(sig_file, "rb") as f:
|
||||||
raw = f.read()
|
raw = f.read()
|
||||||
|
finally:
|
||||||
|
for f in (tbs_file, sig_file):
|
||||||
|
if os.path.exists(f):
|
||||||
|
os.unlink(f)
|
||||||
rb = raw[:48].lstrip(b"\x00") or b"\x00"
|
rb = raw[:48].lstrip(b"\x00") or b"\x00"
|
||||||
sb = raw[48:].lstrip(b"\x00") or b"\x00"
|
sb = raw[48:].lstrip(b"\x00") or b"\x00"
|
||||||
if rb[0] & 0x80: rb = b"\x00" + rb
|
if rb[0] & 0x80: rb = b"\x00" + rb
|
||||||
@ -119,14 +134,19 @@ def build_leaf_cert(cn, sans, days=365):
|
|||||||
content = tbs_full + alg_full + new_sig
|
content = tbs_full + alg_full + new_sig
|
||||||
cl = len(content)
|
cl = len(content)
|
||||||
final = b"\x30\x82" + bytes([cl>>8, cl&0xff]) + content
|
final = b"\x30\x82" + bytes([cl>>8, cl&0xff]) + content
|
||||||
der_file = os.path.join(TMP_DIR, "leaf.der")
|
der_file = _make_temp_file("leaf_", final)
|
||||||
pem_file = os.path.join(TMP_DIR, "leaf.pem")
|
pem_file = _make_temp_file("leaf_pem_")
|
||||||
with open(der_file, "wb") as f: f.write(final)
|
os.unlink(pem_file)
|
||||||
r = subprocess.run(["openssl", "x509", "-inform", "DER", "-outform", "PEM",
|
try:
|
||||||
"-in", der_file, "-out", pem_file],
|
r = subprocess.run(["openssl", "x509", "-inform", "DER", "-outform", "PEM",
|
||||||
capture_output=True, text=True)
|
"-in", der_file, "-out", pem_file],
|
||||||
if r.returncode != 0: return None, r.stderr
|
capture_output=True, text=True)
|
||||||
with open(pem_file) as f: leaf_pem = f.read()
|
if r.returncode != 0: return None, r.stderr
|
||||||
|
with open(pem_file) as f: leaf_pem = f.read()
|
||||||
|
finally:
|
||||||
|
for f in (der_file, pem_file):
|
||||||
|
if os.path.exists(f):
|
||||||
|
os.unlink(f)
|
||||||
key_pem = leaf_key.private_bytes(
|
key_pem = leaf_key.private_bytes(
|
||||||
encoding=serialization.Encoding.PEM,
|
encoding=serialization.Encoding.PEM,
|
||||||
format=serialization.PrivateFormat.PKCS8,
|
format=serialization.PrivateFormat.PKCS8,
|
||||||
|
|||||||
51
tests/test_signing.py
Normal file
51
tests/test_signing.py
Normal file
@ -0,0 +1,51 @@
|
|||||||
|
import os
|
||||||
|
import sys
|
||||||
|
import unittest
|
||||||
|
from unittest import mock
|
||||||
|
|
||||||
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "api"))
|
||||||
|
|
||||||
|
|
||||||
|
class TestSigningSecurity(unittest.TestCase):
|
||||||
|
|
||||||
|
@mock.patch("signing.subprocess.run")
|
||||||
|
@mock.patch("signing._make_temp_file")
|
||||||
|
def test_pin_not_in_cli_args(self, mock_mkstemp, mock_run):
|
||||||
|
"""PIN must NOT appear in subprocess command-line args."""
|
||||||
|
mock_mkstemp.return_value = "/tmp/test_file"
|
||||||
|
mock_run.return_value = mock.MagicMock(returncode=0)
|
||||||
|
import signing
|
||||||
|
signing.sign_tbs_with_yk(b"\x00" * 100, "test-pin-123")
|
||||||
|
cmd = mock_run.call_args[0][0]
|
||||||
|
self.assertNotIn("test-pin-123", cmd, "PIN must not appear in command args")
|
||||||
|
self.assertIn("--pin-source", cmd)
|
||||||
|
self.assertIn("stdin", cmd)
|
||||||
|
self.assertNotIn("--pin", cmd) or cmd.index("--pin-source") < cmd.index("--pin")
|
||||||
|
|
||||||
|
@mock.patch("signing.subprocess.run")
|
||||||
|
def test_temp_files_use_mkstemp(self, mock_run):
|
||||||
|
"""Temp files must use mkstemp, not predictable names."""
|
||||||
|
mock_run.return_value = mock.MagicMock(returncode=0)
|
||||||
|
with mock.patch("signing.tempfile.mkstemp") as mock_mkstemp:
|
||||||
|
mock_mkstemp.return_value = (0, "/tmp/unpredictable_name")
|
||||||
|
import signing
|
||||||
|
try:
|
||||||
|
signing.sign_tbs_with_yk(b"\x00" * 100, "pin")
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
calls = [c[0][1] for c in mock_mkstemp.call_args_list]
|
||||||
|
self.assertTrue(all("/tmp/unpredictable_name" in c for c in calls),
|
||||||
|
"All temp files should use mkstemp")
|
||||||
|
|
||||||
|
def test_no_hardcoded_pin_default(self):
|
||||||
|
"""YK PIN must fail if env var not set."""
|
||||||
|
os.environ.pop("YK_ROOT_PIN", None)
|
||||||
|
os.environ.pop("YK_INT_PIN", None)
|
||||||
|
if "config" in sys.modules:
|
||||||
|
del sys.modules["config"]
|
||||||
|
with self.assertRaises(RuntimeError):
|
||||||
|
__import__("config")
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
unittest.main()
|
||||||
Loading…
x
Reference in New Issue
Block a user