diff --git a/api/signing.py b/api/signing.py index 38ab478..c042f3c 100644 --- a/api/signing.py +++ b/api/signing.py @@ -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.hazmat.primitives import hashes, serialization from cryptography.hazmat.primitives.asymmetric import ec from cryptography.x509.oid import NameOID from config import * -TMP_DIR = "/var/lib/certauth/tmp" os.makedirs(TMP_DIR, exist_ok=True) def get_root_pub_key(): @@ -26,24 +25,40 @@ def der_len(n): elif n < 0x100: return bytes([0x81, n]) 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"): - tbs_file = os.path.join(TMP_DIR, "tbs_sign.der") - sig_file = os.path.join(TMP_DIR, "sig_out.bin") - with open(tbs_file, "wb") as f: - f.write(tbs_bytes) - r = subprocess.run([ - "sudo", "pkcs11-tool", "--module", PKCS11_MODULE, - "--login", "--pin", yk_pin, - "--sign", "--mechanism", "ECDSA-SHA384", - "--token-label", token_label, - "--label", "SIGN key", - "--input-file", tbs_file, - "--output-file", sig_file - ], capture_output=True, text=True) - if r.returncode != 0: - return None, r.stderr - with open(sig_file, "rb") as f: - raw = f.read() + tbs_file = _make_temp_file("tbs_", tbs_bytes) + sig_file = _make_temp_file("sig_") + os.unlink(sig_file) + try: + r = subprocess.run([ + "sudo", "pkcs11-tool", "--module", PKCS11_MODULE, + "--login", "--pin-source", "stdin", + "--sign", "--mechanism", "ECDSA-SHA384", + "--token-label", token_label, + "--label", "SIGN key", + "--input-file", tbs_file, + "--output-file", sig_file + ], input=yk_pin, capture_output=True, text=True) + if r.returncode != 0: + return None, r.stderr + with open(sig_file, "rb") as f: + 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" sb = raw[48:].lstrip(b"\x00") or b"\x00" 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 cl = len(content) final = b"\x30\x82" + bytes([cl>>8, cl&0xff]) + content - der_file = os.path.join(TMP_DIR, "leaf.der") - pem_file = os.path.join(TMP_DIR, "leaf.pem") - with open(der_file, "wb") as f: f.write(final) - r = subprocess.run(["openssl", "x509", "-inform", "DER", "-outform", "PEM", - "-in", der_file, "-out", pem_file], - capture_output=True, text=True) - if r.returncode != 0: return None, r.stderr - with open(pem_file) as f: leaf_pem = f.read() + der_file = _make_temp_file("leaf_", final) + pem_file = _make_temp_file("leaf_pem_") + os.unlink(pem_file) + try: + r = subprocess.run(["openssl", "x509", "-inform", "DER", "-outform", "PEM", + "-in", der_file, "-out", pem_file], + capture_output=True, text=True) + 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( encoding=serialization.Encoding.PEM, format=serialization.PrivateFormat.PKCS8, diff --git a/tests/test_signing.py b/tests/test_signing.py new file mode 100644 index 0000000..2572b46 --- /dev/null +++ b/tests/test_signing.py @@ -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()