Initial commit
This commit is contained in:
834
main.py
Normal file
834
main.py
Normal file
@@ -0,0 +1,834 @@
|
||||
import argparse
|
||||
import os
|
||||
import secrets
|
||||
import select
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
import zipfile
|
||||
from datetime import datetime
|
||||
|
||||
from smartcard.Exceptions import CardConnectionException, NoCardException
|
||||
from smartcard.System import readers
|
||||
from smartcard.util import toHexString
|
||||
|
||||
from profiles import AppletProfile, DefaultProfile, get_profile
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Constants
|
||||
# ---------------------------------------------------------------------------
|
||||
DEFAULT_KEY = "404142434445464748494A4B4C4D4E4F"
|
||||
|
||||
# GET DATA for card UID (works on most contactless readers)
|
||||
GET_UID_APDU = [0xFF, 0xCA, 0x00, 0x00, 0x00]
|
||||
|
||||
GP_JAR = "gp.jar"
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# CAP file parsing
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def extract_applet_aids(cap_file: str) -> list[list[int]]:
|
||||
"""
|
||||
Extract applet AIDs from a CAP file's Applet.cap component.
|
||||
|
||||
Returns a list of AIDs as int-lists (ready for SELECT APDUs).
|
||||
Returns an empty list on any parse error.
|
||||
"""
|
||||
try:
|
||||
with zipfile.ZipFile(cap_file, "r") as zf:
|
||||
applet_entries = [n for n in zf.namelist() if n.endswith("Applet.cap")]
|
||||
if not applet_entries:
|
||||
return []
|
||||
data = zf.read(applet_entries[0])
|
||||
except (zipfile.BadZipFile, OSError):
|
||||
return []
|
||||
|
||||
if len(data) < 4 or data[0] != 0x03:
|
||||
return []
|
||||
|
||||
count = data[3]
|
||||
aids: list[list[int]] = []
|
||||
offset = 4
|
||||
for _ in range(count):
|
||||
if offset >= len(data):
|
||||
break
|
||||
aid_len = data[offset]
|
||||
offset += 1
|
||||
if offset + aid_len + 2 > len(data):
|
||||
break
|
||||
aid = list(data[offset : offset + aid_len])
|
||||
offset += aid_len + 2 # skip install_method_offset (u2)
|
||||
aids.append(aid)
|
||||
return aids
|
||||
|
||||
|
||||
def _build_cap_profiles(
|
||||
cap_files: list[str],
|
||||
) -> dict[str, list[tuple[list[int], AppletProfile]]]:
|
||||
"""
|
||||
For each CAP file, extract applet AIDs and resolve their profiles.
|
||||
|
||||
Returns {cap_path: [(aid, profile), ...]}.
|
||||
"""
|
||||
result: dict[str, list[tuple[list[int], AppletProfile]]] = {}
|
||||
for cap in cap_files:
|
||||
aids = extract_applet_aids(cap)
|
||||
entries = []
|
||||
for aid in aids:
|
||||
profile = get_profile(aid)
|
||||
if profile is None:
|
||||
profile = DefaultProfile(aid)
|
||||
entries.append((aid, profile))
|
||||
result[cap] = entries
|
||||
return result
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def validate_hex_key(value: str, name: str) -> str:
|
||||
"""Validate that *value* is exactly 32 hex characters (16 bytes)."""
|
||||
stripped = value.strip()
|
||||
if len(stripped) != 32:
|
||||
raise argparse.ArgumentTypeError(
|
||||
f"{name}: expected 32 hex characters (16 bytes), got {len(stripped)}"
|
||||
)
|
||||
try:
|
||||
bytes.fromhex(stripped)
|
||||
except ValueError:
|
||||
raise argparse.ArgumentTypeError(
|
||||
f"{name}: invalid hex characters in '{stripped}'"
|
||||
)
|
||||
return stripped.upper()
|
||||
|
||||
|
||||
def hex_key_type(value: str) -> str:
|
||||
"""argparse *type* callback for key arguments."""
|
||||
return validate_hex_key(value, "key")
|
||||
|
||||
|
||||
def generate_random_key() -> str:
|
||||
"""Return a cryptographically random 16-byte key as a 32-char hex string."""
|
||||
return secrets.token_hex(16).upper()
|
||||
|
||||
|
||||
def resolve_keys(args: argparse.Namespace) -> tuple[str, str, str]:
|
||||
"""Return (enc, mac, dek) from the parsed CLI arguments for *current* keys."""
|
||||
individual = [args.key_enc, args.key_mac, args.key_dek]
|
||||
provided = [k for k in individual if k is not None]
|
||||
|
||||
if len(provided) not in (0, 3):
|
||||
_die("When using --key-enc / --key-mac / --key-dek you must supply all three.")
|
||||
|
||||
if provided:
|
||||
return (args.key_enc, args.key_mac, args.key_dek)
|
||||
|
||||
base = args.key or DEFAULT_KEY
|
||||
return (base, base, base)
|
||||
|
||||
|
||||
def resolve_new_keys(args: argparse.Namespace) -> tuple[str, str, str] | None:
|
||||
"""Return (enc, mac, dek) for the *new* keys, or None if no key change requested."""
|
||||
if not args.change_keys and not args.randomize_keys:
|
||||
return None
|
||||
|
||||
if args.randomize_keys:
|
||||
return (generate_random_key(), generate_random_key(), generate_random_key())
|
||||
|
||||
individual = [args.new_key_enc, args.new_key_mac, args.new_key_dek]
|
||||
provided = [k for k in individual if k is not None]
|
||||
|
||||
if provided and len(provided) != 3:
|
||||
_die("When using --new-key-enc / --new-key-mac / --new-key-dek you must supply all three.")
|
||||
|
||||
if provided:
|
||||
return (args.new_key_enc, args.new_key_mac, args.new_key_dek)
|
||||
|
||||
if args.new_key:
|
||||
return (args.new_key, args.new_key, args.new_key)
|
||||
|
||||
if args.change_keys:
|
||||
_die("--change-keys requires either --new-key, all three --new-key-enc/mac/dek, or --randomize-keys.")
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def _die(msg: str) -> None:
|
||||
"""Print an error message and exit."""
|
||||
print(f"\n Error: {msg}\n", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
def _info(msg: str) -> None:
|
||||
print(f" {msg}")
|
||||
|
||||
|
||||
def _success(msg: str) -> None:
|
||||
print(f" [OK] {msg}")
|
||||
|
||||
|
||||
def _warn(msg: str) -> None:
|
||||
print(f" [!] {msg}", file=sys.stderr)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Card detection (pyscard)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _read_uid_quick(reader) -> str | None:
|
||||
"""Try to read the UID of whatever card is on the reader. Returns None if no card."""
|
||||
try:
|
||||
conn = reader.createConnection()
|
||||
conn.connect()
|
||||
response, sw1, sw2 = conn.transmit(GET_UID_APDU)
|
||||
conn.disconnect()
|
||||
if sw1 == 0x90 and sw2 == 0x00:
|
||||
return toHexString(response).replace(" ", "")
|
||||
except (CardConnectionException, NoCardException):
|
||||
pass
|
||||
return None
|
||||
|
||||
|
||||
|
||||
def _resolve_reader(reader_name: str | None = None):
|
||||
"""Find and return the smartcard reader to use."""
|
||||
available = readers()
|
||||
if not available:
|
||||
_die("No smartcard readers found. Is a reader connected?")
|
||||
|
||||
if reader_name:
|
||||
matches = [r for r in available if reader_name.lower() in str(r).lower()]
|
||||
if not matches:
|
||||
_die(
|
||||
f"Reader '{reader_name}' not found. Available readers:\n"
|
||||
+ "\n".join(f" - {r}" for r in available)
|
||||
)
|
||||
return matches[0]
|
||||
return available[0]
|
||||
|
||||
|
||||
def _stdin_ready() -> bool:
|
||||
"""Return True if the user has pressed Enter (non-blocking)."""
|
||||
ready, _, _ = select.select([sys.stdin], [], [], 0)
|
||||
if ready:
|
||||
sys.stdin.readline() # consume the newline
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def detect_card(reader, previous_uid: str | None = None):
|
||||
"""
|
||||
Wait for a new card and return (connection, uid_hex), or None if
|
||||
the user presses Enter to end the batch.
|
||||
|
||||
If *previous_uid* is set, keeps polling until a different card appears.
|
||||
"""
|
||||
# Flush any buffered stdin so stray keypresses during card processing
|
||||
# don't immediately end the batch.
|
||||
while select.select([sys.stdin], [], [], 0)[0]:
|
||||
sys.stdin.readline()
|
||||
|
||||
_info("Waiting for card... (press Enter to end batch)")
|
||||
|
||||
# Poll until a *new* card is present (different from previous_uid).
|
||||
poll_count = 0
|
||||
while True:
|
||||
if _stdin_ready():
|
||||
return None
|
||||
|
||||
# Periodically verify the reader is still connected (~every 5 s).
|
||||
poll_count += 1
|
||||
if poll_count % 30 == 0:
|
||||
try:
|
||||
available = readers()
|
||||
if str(reader) not in [str(r) for r in available]:
|
||||
_warn("Reader disconnected.")
|
||||
return None
|
||||
except Exception:
|
||||
_warn("Reader disconnected.")
|
||||
return None
|
||||
|
||||
uid_hex = _read_uid_quick(reader)
|
||||
|
||||
if uid_hex is not None:
|
||||
if previous_uid is None or uid_hex != previous_uid:
|
||||
break
|
||||
# Same card still on reader — keep waiting
|
||||
time.sleep(0.15)
|
||||
|
||||
# We know a new card is present. Connect fresh for real APDUs.
|
||||
# The quick-read disconnected (dropping RF briefly), so retry if needed.
|
||||
connection = None
|
||||
for _ in range(20):
|
||||
try:
|
||||
connection = reader.createConnection()
|
||||
connection.connect()
|
||||
break
|
||||
except (CardConnectionException, NoCardException):
|
||||
time.sleep(0.15)
|
||||
if connection is None:
|
||||
_warn("Card vanished before we could connect. Retrying...")
|
||||
return "retry"
|
||||
|
||||
# Re-read UID on this connection (the _read_uid_quick one disconnected).
|
||||
try:
|
||||
response, sw1, sw2 = connection.transmit(GET_UID_APDU)
|
||||
if sw1 == 0x90 and sw2 == 0x00:
|
||||
uid_hex = toHexString(response).replace(" ", "")
|
||||
except (CardConnectionException, NoCardException):
|
||||
pass # keep the uid_hex from the quick read
|
||||
|
||||
return connection, uid_hex
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# gp.jar interaction
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _find_gp_jar() -> str:
|
||||
"""Locate gp.jar — check current dir, then PATH."""
|
||||
local = os.path.join(os.getcwd(), GP_JAR)
|
||||
if os.path.isfile(local):
|
||||
return local
|
||||
|
||||
# Check if 'gp' wrapper script or alias exists on PATH
|
||||
gp_on_path = shutil.which("gp")
|
||||
if gp_on_path:
|
||||
return gp_on_path
|
||||
|
||||
gp_jar_on_path = shutil.which(GP_JAR)
|
||||
if gp_jar_on_path:
|
||||
return gp_jar_on_path
|
||||
|
||||
_die(
|
||||
f"{GP_JAR} not found.\n"
|
||||
" Place gp.jar in the current directory, or add it to your PATH.\n"
|
||||
" Download from: https://github.com/martinpaljak/GlobalPlatformPro/releases"
|
||||
)
|
||||
|
||||
|
||||
def _check_java() -> None:
|
||||
"""Ensure a Java runtime is available."""
|
||||
if shutil.which("java") is None:
|
||||
_die("Java runtime not found. Please install Java (JRE 8+).")
|
||||
|
||||
|
||||
def _build_key_flags(keys: tuple[str, str, str], prefix: str = "") -> list[str]:
|
||||
"""
|
||||
Build gp.jar key flags.
|
||||
|
||||
*prefix* is '' for current keys (-key*) or 'lock' for new keys (--lock*).
|
||||
"""
|
||||
enc, mac, dek = keys
|
||||
if prefix:
|
||||
flag_single = f"--{prefix}"
|
||||
flag_enc = f"--{prefix}-enc"
|
||||
flag_mac = f"--{prefix}-mac"
|
||||
flag_dek = f"--{prefix}-dek"
|
||||
else:
|
||||
flag_single = "--key"
|
||||
flag_enc = "--key-enc"
|
||||
flag_mac = "--key-mac"
|
||||
flag_dek = "--key-dek"
|
||||
|
||||
if enc == mac == dek:
|
||||
return [flag_single, enc]
|
||||
if prefix:
|
||||
# gp.jar requires --lock as a base even when individual lock keys differ.
|
||||
# Set the base to enc, then override mac and dek.
|
||||
return [flag_single, enc, flag_mac, mac, flag_dek, dek]
|
||||
# For current keys, use individual --key-enc/mac/dek flags.
|
||||
return [flag_enc, enc, flag_mac, mac, flag_dek, dek]
|
||||
|
||||
|
||||
def run_gp(gp_args: list[str], keys: tuple[str, str, str]) -> subprocess.CompletedProcess:
|
||||
"""
|
||||
Run gp.jar with the given arguments and key flags.
|
||||
|
||||
Raises SystemExit with a friendly message on failure.
|
||||
"""
|
||||
gp_path = _find_gp_jar()
|
||||
|
||||
# If gp_path is a .jar, run via java -jar; otherwise assume it's a wrapper
|
||||
if gp_path.endswith(".jar"):
|
||||
cmd = ["java", "-jar", gp_path]
|
||||
else:
|
||||
cmd = [gp_path]
|
||||
|
||||
cmd += _build_key_flags(keys)
|
||||
cmd += gp_args
|
||||
|
||||
try:
|
||||
result = subprocess.run(
|
||||
cmd,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=120,
|
||||
)
|
||||
except subprocess.TimeoutExpired:
|
||||
raise RuntimeError("gp.jar timed out after 120 seconds.")
|
||||
except FileNotFoundError:
|
||||
raise RuntimeError("Failed to launch gp.jar. Is Java installed?")
|
||||
|
||||
if result.returncode != 0:
|
||||
stderr = result.stderr.strip() or result.stdout.strip()
|
||||
raise RuntimeError(f"gp.jar failed (exit {result.returncode}):\n {stderr}")
|
||||
|
||||
return result
|
||||
|
||||
|
||||
def install_cap(cap_file: str, keys: tuple[str, str, str]) -> None:
|
||||
"""Install a CAP file onto the card via gp.jar."""
|
||||
_info(f"Installing {os.path.basename(cap_file)} ...")
|
||||
try:
|
||||
run_gp(["--install", cap_file], keys)
|
||||
_success(f"Installed {os.path.basename(cap_file)}")
|
||||
except RuntimeError as exc:
|
||||
raise RuntimeError(f"Failed to install {os.path.basename(cap_file)}: {exc}")
|
||||
|
||||
|
||||
def change_keys(current_keys: tuple[str, str, str], new_keys: tuple[str, str, str]) -> None:
|
||||
"""Change the card's GP keys from *current_keys* to *new_keys*."""
|
||||
_info("Changing card keys...")
|
||||
lock_flags = _build_key_flags(new_keys, prefix="lock")
|
||||
try:
|
||||
run_gp(lock_flags, current_keys)
|
||||
_success("Keys changed successfully.")
|
||||
except RuntimeError as exc:
|
||||
raise RuntimeError(f"Key change failed: {exc}")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Card APDU helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def select_aid(connection, aid: list[int]) -> bool:
|
||||
"""
|
||||
SELECT an applet by AID.
|
||||
|
||||
Returns True if the applet responds with SW 90 00, False otherwise.
|
||||
"""
|
||||
select_apdu = [0x00, 0xA4, 0x04, 0x00, len(aid)] + aid
|
||||
try:
|
||||
response, sw1, sw2 = connection.transmit(select_apdu)
|
||||
except (CardConnectionException, NoCardException):
|
||||
return False
|
||||
return (sw1, sw2) == (0x90, 0x00)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Logging
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _write_log(log_file: str, text: str) -> None:
|
||||
"""Append text to the log file, warning (not crashing) on failure."""
|
||||
try:
|
||||
with open(log_file, "a") as f:
|
||||
f.write(text)
|
||||
except OSError as exc:
|
||||
_warn(f"Failed to write log: {exc}")
|
||||
|
||||
|
||||
def write_log_header(log_file: str, start_time: datetime) -> None:
|
||||
"""Write the session header to the log file."""
|
||||
sep = "=" * 60
|
||||
_write_log(log_file,
|
||||
f"{sep}\n"
|
||||
f"Batch Session Started: {start_time.strftime('%Y-%m-%d %H:%M:%S')}\n"
|
||||
f"{sep}\n\n"
|
||||
)
|
||||
|
||||
|
||||
def write_log_entry(
|
||||
log_file: str,
|
||||
card_number: int,
|
||||
uid: str,
|
||||
validation_results: dict[str, dict[str, str]],
|
||||
status: str,
|
||||
) -> None:
|
||||
"""Append a single card entry to the log file."""
|
||||
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||||
lines = [
|
||||
f"--- Card #{card_number} ---\n",
|
||||
f" Time: {now}\n",
|
||||
f" UID: {uid}\n",
|
||||
]
|
||||
if validation_results:
|
||||
for profile_name, result in validation_results.items():
|
||||
lines.append(f" [{profile_name}]\n")
|
||||
for key, value in result.items():
|
||||
lines.append(f" {key}: {value}\n")
|
||||
lines.append(f" Status: {status}\n\n")
|
||||
_write_log(log_file, "".join(lines))
|
||||
|
||||
|
||||
def write_log_footer(log_file: str, total: int) -> None:
|
||||
"""Write the session footer to the log file."""
|
||||
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||||
sep = "=" * 60
|
||||
_write_log(log_file,
|
||||
f"{'-' * 60}\n"
|
||||
f"Batch Complete: {now}\n"
|
||||
f"Total Cards Processed: {total}\n"
|
||||
f"{sep}\n\n"
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# CLI
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def build_parser() -> argparse.ArgumentParser:
|
||||
parser = argparse.ArgumentParser(
|
||||
prog="j3r180-batch",
|
||||
description="Batch provisioning tool for JavaCard smartcards (Tesla key cards).",
|
||||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||||
epilog=(
|
||||
"examples:\n"
|
||||
" %(prog)s --cap-file app.cap\n"
|
||||
" %(prog)s --cap-file app.cap --randomize-keys\n"
|
||||
" %(prog)s --cap-file app.cap --change-keys --new-key AABBCCDDEEFF00112233445566778899\n"
|
||||
" %(prog)s --key 00112233445566778899AABBCCDDEEFF --cap-file a.cap b.cap\n"
|
||||
),
|
||||
)
|
||||
|
||||
# --- Current key flags ---
|
||||
key_group = parser.add_argument_group("current card keys (for GP authentication)")
|
||||
key_group.add_argument(
|
||||
"-k", "--key",
|
||||
type=hex_key_type,
|
||||
default=None,
|
||||
metavar="HEX",
|
||||
help=f"single key for ENC/MAC/DEK (default: {DEFAULT_KEY})",
|
||||
)
|
||||
key_group.add_argument("--key-enc", type=hex_key_type, default=None, metavar="HEX", help="ENC key")
|
||||
key_group.add_argument("--key-mac", type=hex_key_type, default=None, metavar="HEX", help="MAC key")
|
||||
key_group.add_argument("--key-dek", type=hex_key_type, default=None, metavar="HEX", help="DEK key")
|
||||
|
||||
# --- CAP file ---
|
||||
parser.add_argument(
|
||||
"--cap-file",
|
||||
nargs="+",
|
||||
metavar="FILE",
|
||||
help="one or more .cap files to install",
|
||||
)
|
||||
|
||||
# --- Key change flags ---
|
||||
lock_group = parser.add_argument_group("key change options")
|
||||
lock_group.add_argument(
|
||||
"--change-keys",
|
||||
action="store_true",
|
||||
help="change the card's GP keys after installation",
|
||||
)
|
||||
lock_group.add_argument("--new-key", type=hex_key_type, default=None, metavar="HEX", help="new single key for ENC/MAC/DEK")
|
||||
lock_group.add_argument("--new-key-enc", type=hex_key_type, default=None, metavar="HEX", help="new ENC key")
|
||||
lock_group.add_argument("--new-key-mac", type=hex_key_type, default=None, metavar="HEX", help="new MAC key")
|
||||
lock_group.add_argument("--new-key-dek", type=hex_key_type, default=None, metavar="HEX", help="new DEK key")
|
||||
lock_group.add_argument(
|
||||
"--randomize-keys",
|
||||
action="store_true",
|
||||
help="generate random keys and lock the card (keys are NOT recorded)",
|
||||
)
|
||||
|
||||
# --- Other ---
|
||||
parser.add_argument(
|
||||
"--log-file",
|
||||
default=None,
|
||||
metavar="PATH",
|
||||
help="override log file path (default: auto-generated from timestamp + params)",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--reader",
|
||||
default=None,
|
||||
metavar="NAME",
|
||||
help="smartcard reader name or substring (auto-detect if omitted)",
|
||||
)
|
||||
|
||||
return parser
|
||||
|
||||
|
||||
def validate_args(args: argparse.Namespace) -> None:
|
||||
"""Perform cross-field validation that argparse can't handle alone."""
|
||||
# --key and --key-enc/mac/dek are mutually exclusive
|
||||
if args.key and any([args.key_enc, args.key_mac, args.key_dek]):
|
||||
_die("Cannot combine --key with --key-enc/mac/dek. Use one or the other.")
|
||||
|
||||
# Same for new-key variants
|
||||
if args.new_key and any([args.new_key_enc, args.new_key_mac, args.new_key_dek]):
|
||||
_die("Cannot combine --new-key with --new-key-enc/mac/dek. Use one or the other.")
|
||||
|
||||
# CAP file validation
|
||||
if args.cap_file:
|
||||
for path in args.cap_file:
|
||||
if not path.lower().endswith(".cap"):
|
||||
_die(f"'{path}' does not look like a .cap file.")
|
||||
if not os.path.isfile(path):
|
||||
_die(f"CAP file not found: {path}")
|
||||
|
||||
# --randomize-keys implies --change-keys
|
||||
if args.randomize_keys:
|
||||
args.change_keys = True
|
||||
|
||||
# --new-key* implies --change-keys
|
||||
if any([args.new_key, args.new_key_enc, args.new_key_mac, args.new_key_dek]):
|
||||
args.change_keys = True
|
||||
|
||||
# --randomize-keys is mutually exclusive with explicit new keys
|
||||
if args.randomize_keys and any([args.new_key, args.new_key_enc, args.new_key_mac, args.new_key_dek]):
|
||||
_die("--randomize-keys cannot be combined with --new-key / --new-key-enc/mac/dek.")
|
||||
|
||||
# Must have something to do
|
||||
if not args.cap_file and not args.change_keys:
|
||||
_die("Nothing to do. Specify --cap-file and/or key change flags.")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Main loop
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def process_card(
|
||||
args: argparse.Namespace,
|
||||
reader,
|
||||
current_keys: tuple[str, str, str],
|
||||
new_keys: tuple[str, str, str] | None,
|
||||
connection,
|
||||
uid: str,
|
||||
cap_profiles: dict[str, list[tuple[list[int], AppletProfile]]],
|
||||
) -> tuple[str, dict[str, dict[str, str]]]:
|
||||
"""
|
||||
Process a single card that has already been detected.
|
||||
|
||||
Returns (uid, validation_results) where validation_results maps
|
||||
profile name → {field: value} from each profile's validate().
|
||||
"""
|
||||
_success(f"Card detected — UID: {uid}")
|
||||
|
||||
gp_ran = False
|
||||
disconnected = False
|
||||
|
||||
# --- Per-CAP install: SELECT each applet AID to check presence ---
|
||||
if args.cap_file:
|
||||
for cap in args.cap_file:
|
||||
basename = os.path.basename(cap)
|
||||
entries = cap_profiles.get(cap, [])
|
||||
aids = [aid for aid, _ in entries]
|
||||
|
||||
# If we can extract AIDs, check if all applets are already present.
|
||||
already = False
|
||||
if aids and not disconnected:
|
||||
already = all(select_aid(connection, aid) for aid in aids)
|
||||
|
||||
if already:
|
||||
_info(f"{basename} already installed — skipping.")
|
||||
continue
|
||||
|
||||
# Need gp.jar — disconnect pyscard first (once).
|
||||
if not disconnected:
|
||||
try:
|
||||
connection.disconnect()
|
||||
except Exception:
|
||||
pass
|
||||
disconnected = True
|
||||
|
||||
install_cap(cap, current_keys)
|
||||
gp_ran = True
|
||||
|
||||
# --- Change keys ---
|
||||
if new_keys:
|
||||
if not disconnected:
|
||||
try:
|
||||
connection.disconnect()
|
||||
except Exception:
|
||||
pass
|
||||
disconnected = True
|
||||
|
||||
if args.randomize_keys:
|
||||
new_keys = (generate_random_key(), generate_random_key(), generate_random_key())
|
||||
try:
|
||||
change_keys(current_keys, new_keys)
|
||||
gp_ran = True
|
||||
except RuntimeError as exc:
|
||||
# Best-effort when applet is already provisioned.
|
||||
if args.cap_file and not gp_ran:
|
||||
_warn(f"Key change failed (non-fatal): {exc}")
|
||||
else:
|
||||
raise
|
||||
|
||||
# --- Reconnect if needed ---
|
||||
if gp_ran:
|
||||
# gp.jar took the reader; reconnect via pyscard (RF field dropped briefly).
|
||||
connection = None
|
||||
for _ in range(40): # ~6 seconds max
|
||||
try:
|
||||
connection = reader.createConnection()
|
||||
connection.connect()
|
||||
break
|
||||
except (CardConnectionException, NoCardException):
|
||||
time.sleep(0.15)
|
||||
if connection is None:
|
||||
_warn("Card not detected after GP operations. Was it removed?")
|
||||
return uid, {}
|
||||
elif disconnected:
|
||||
try:
|
||||
connection = reader.createConnection()
|
||||
connection.connect()
|
||||
except (CardConnectionException, NoCardException):
|
||||
_warn("Card not detected. Was it removed?")
|
||||
return uid, {}
|
||||
# else: connection is still live from detect_card — use it directly.
|
||||
|
||||
# --- Validate installed applets via profiles ---
|
||||
validation_results: dict[str, dict[str, str]] = {}
|
||||
if args.cap_file:
|
||||
for cap in args.cap_file:
|
||||
for aid, profile in cap_profiles.get(cap, []):
|
||||
result = profile.validate(connection)
|
||||
validation_results[profile.name] = result
|
||||
|
||||
if result.get("installed") == "yes":
|
||||
if "error" in result:
|
||||
_warn(f"{profile.name}: installed but {result['error']}")
|
||||
else:
|
||||
_success(f"{profile.name}: validated")
|
||||
for key, value in result.items():
|
||||
if key != "installed":
|
||||
_info(f" {key}: {value}")
|
||||
else:
|
||||
_warn(f"{profile.name}: {result.get('error', 'validation failed')}")
|
||||
|
||||
try:
|
||||
connection.disconnect()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
return uid, validation_results
|
||||
|
||||
|
||||
def main() -> None:
|
||||
parser = build_parser()
|
||||
args = parser.parse_args()
|
||||
validate_args(args)
|
||||
|
||||
# Resolve keys (validates partial key combos)
|
||||
current_keys = resolve_keys(args)
|
||||
new_keys = resolve_new_keys(args)
|
||||
|
||||
# Pre-flight checks
|
||||
_check_java()
|
||||
_find_gp_jar()
|
||||
|
||||
# Resolve CAP → profile mappings
|
||||
cap_profiles: dict[str, list[tuple[list[int], AppletProfile]]] = {}
|
||||
if args.cap_file:
|
||||
cap_profiles = _build_cap_profiles(args.cap_file)
|
||||
|
||||
reader = _resolve_reader(args.reader)
|
||||
start_time = datetime.now()
|
||||
card_count = 0
|
||||
|
||||
# Build default log filename: timestamp + non-key params
|
||||
if args.log_file is None:
|
||||
parts = [start_time.strftime("%Y%m%d_%H%M%S")]
|
||||
if args.cap_file:
|
||||
parts.extend(
|
||||
os.path.splitext(os.path.basename(f))[0] for f in args.cap_file
|
||||
)
|
||||
if args.randomize_keys:
|
||||
parts.append("randomize")
|
||||
elif args.change_keys:
|
||||
parts.append("rekey")
|
||||
args.log_file = "_".join(parts) + ".log"
|
||||
|
||||
# Print banner
|
||||
print()
|
||||
print("=" * 50)
|
||||
print(" j3r180 Batch Provisioning Tool")
|
||||
print("=" * 50)
|
||||
_info(f"Session started: {start_time.strftime('%Y-%m-%d %H:%M:%S')}")
|
||||
_info(f"Reader: {reader}")
|
||||
if args.cap_file:
|
||||
for cap in args.cap_file:
|
||||
profiles = [p.name for _, p in cap_profiles.get(cap, [])]
|
||||
label = ", ".join(profiles) if profiles else "no profile"
|
||||
_info(f"CAP: {os.path.basename(cap)} -> {label}")
|
||||
if args.randomize_keys:
|
||||
_info("Keys will be RANDOMIZED (not recorded).")
|
||||
elif new_keys:
|
||||
_info("Keys will be changed.")
|
||||
_info(f"Log file: {args.log_file}")
|
||||
print()
|
||||
|
||||
# Validate log file is writable before starting the batch.
|
||||
try:
|
||||
with open(args.log_file, "a") as f:
|
||||
pass
|
||||
except OSError as exc:
|
||||
_die(f"Cannot write to log file '{args.log_file}': {exc}")
|
||||
|
||||
write_log_header(args.log_file, start_time)
|
||||
|
||||
last_uid = None
|
||||
|
||||
try:
|
||||
while True:
|
||||
result = detect_card(reader, previous_uid=last_uid)
|
||||
if result is None:
|
||||
# User pressed Enter
|
||||
break
|
||||
if result == "retry":
|
||||
# Card vanished mid-detect, loop back
|
||||
continue
|
||||
|
||||
connection, uid = result
|
||||
card_count += 1
|
||||
print(f"--- Card #{card_count} ---")
|
||||
|
||||
try:
|
||||
uid, validation_results = process_card(
|
||||
args, reader, current_keys, new_keys, connection, uid,
|
||||
cap_profiles,
|
||||
)
|
||||
write_log_entry(args.log_file, card_count, uid, validation_results, "OK")
|
||||
last_uid = uid
|
||||
print()
|
||||
_success(f"Card #{card_count} complete. ({card_count} total)")
|
||||
except (RuntimeError, CardConnectionException, NoCardException) as exc:
|
||||
_warn(f"Card #{card_count} failed: {exc}")
|
||||
print()
|
||||
retry = input(" Retry this card? [Y/n]: ").strip().lower()
|
||||
if retry in ("", "y", "yes"):
|
||||
card_count -= 1
|
||||
# Card must leave the RF field to reset after a failed
|
||||
# operation — wait for removal before re-detecting.
|
||||
_info("Remove card from reader to reset it...")
|
||||
while _read_uid_quick(reader) is not None:
|
||||
time.sleep(0.15)
|
||||
_info("Card removed. Re-present when ready.")
|
||||
last_uid = None # allow same UID to be picked up again
|
||||
else:
|
||||
write_log_entry(
|
||||
args.log_file, card_count,
|
||||
uid or "UNKNOWN", {}, f"FAILED: {exc}",
|
||||
)
|
||||
last_uid = uid
|
||||
|
||||
print()
|
||||
|
||||
except KeyboardInterrupt:
|
||||
print("\n")
|
||||
_warn("Interrupted by user.")
|
||||
|
||||
# Final summary
|
||||
write_log_footer(args.log_file, card_count)
|
||||
print()
|
||||
print("-" * 50)
|
||||
_info(f"Session complete. {card_count} card(s) processed.")
|
||||
_info(f"Log written to: {os.path.abspath(args.log_file)}")
|
||||
print()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user