refactor: unifier le calcul d'empreinte des cles SSH
generate_key() calculait une vraie empreinte SHA256, mais import_key() prenait juste pub_text.split()[1][:16] - un fragment de base64, pas une empreinte. - compute_fingerprint(public_key_line) : fonction unique (format "SHA256:xxxx...", comme ssh-keygen -lf), utilisee par generate_key() et import_key(). - KeyManager.load() corrige silencieusement les entrees existantes dont l'empreinte ne correspond pas a compute_fingerprint(public_key_text) (cas des cles importees avant ce correctif), et resauvegarde si necessaire. Ce commit branche aussi dans KeyManager les deux mecanismes ajoutes par les commits precedents : generate_key() ecrit desormais via app.storage.json_store (schema_version=1, .bak, ecriture atomique) et protege la cle privee avec app.security.key_protection.protect_private_key() a la place du chmod(0o600) precedent. Verifie par tests/test_atomic_storage.py (recalcul d'une empreinte tronquee au chargement) et tests/test_keys.py (format SHA256: pour toutes les cles generees). Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
@@ -1,7 +1,8 @@
|
|||||||
"""Gestion des cles SSH : generation, import, export, registre local."""
|
"""Gestion des cles SSH : generation, import, export, registre local."""
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
import json
|
import base64
|
||||||
|
import hashlib
|
||||||
import shutil
|
import shutil
|
||||||
import uuid
|
import uuid
|
||||||
from dataclasses import dataclass, field, asdict
|
from dataclasses import dataclass, field, asdict
|
||||||
@@ -12,8 +13,31 @@ from cryptography.hazmat.primitives import serialization
|
|||||||
from cryptography.hazmat.primitives.asymmetric import rsa, ed25519, ec
|
from cryptography.hazmat.primitives.asymmetric import rsa, ed25519, ec
|
||||||
|
|
||||||
from app.paths import DATA_DIR # noqa: E402 (chemins valables aussi en .exe)
|
from app.paths import DATA_DIR # noqa: E402 (chemins valables aussi en .exe)
|
||||||
|
from app.storage.json_store import load_json_document, save_json_document
|
||||||
|
from app.security.key_protection import protect_private_key
|
||||||
|
|
||||||
KEYS_FILE = DATA_DIR / "keys.json"
|
KEYS_FILE = DATA_DIR / "keys.json"
|
||||||
DEFAULT_KEY_DIR = DATA_DIR / "keys"
|
DEFAULT_KEY_DIR = DATA_DIR / "keys"
|
||||||
|
SCHEMA_VERSION = 1
|
||||||
|
|
||||||
|
|
||||||
|
def compute_fingerprint(public_key_line: str) -> str:
|
||||||
|
"""Empreinte SHA256 canonique (format `SHA256:xxxx...`, comme `ssh-keygen -lf`),
|
||||||
|
a partir d'une ligne de cle publique OpenSSH ("type base64 [commentaire]").
|
||||||
|
|
||||||
|
Utilisee pour toute cle, qu'elle soit generee, importee ou relue depuis le
|
||||||
|
registre : jamais un simple fragment du base64, qui ne serait pas un vrai
|
||||||
|
condense cryptographique de la cle."""
|
||||||
|
parts = (public_key_line or "").split()
|
||||||
|
if len(parts) < 2:
|
||||||
|
return ""
|
||||||
|
try:
|
||||||
|
raw = base64.b64decode(parts[1])
|
||||||
|
except (ValueError, TypeError):
|
||||||
|
return ""
|
||||||
|
digest = hashlib.sha256(raw).digest()
|
||||||
|
return "SHA256:" + base64.b64encode(digest).decode("utf-8").rstrip("=")
|
||||||
|
|
||||||
|
|
||||||
KEY_TYPE_ED25519 = "ed25519"
|
KEY_TYPE_ED25519 = "ed25519"
|
||||||
KEY_TYPE_RSA = "rsa"
|
KEY_TYPE_RSA = "rsa"
|
||||||
@@ -104,26 +128,39 @@ class KeyManager:
|
|||||||
def __init__(self, registry_path: Path = KEYS_FILE):
|
def __init__(self, registry_path: Path = KEYS_FILE):
|
||||||
self.registry_path = registry_path
|
self.registry_path = registry_path
|
||||||
self.keys: list[KeyEntry] = []
|
self.keys: list[KeyEntry] = []
|
||||||
|
self.load_warning: str | None = None
|
||||||
self.load()
|
self.load()
|
||||||
|
|
||||||
# ---------- registre ----------
|
# ---------- registre ----------
|
||||||
def load(self) -> None:
|
def load(self) -> None:
|
||||||
DATA_DIR.mkdir(parents=True, exist_ok=True)
|
|
||||||
DEFAULT_KEY_DIR.mkdir(parents=True, exist_ok=True)
|
DEFAULT_KEY_DIR.mkdir(parents=True, exist_ok=True)
|
||||||
if not self.registry_path.exists():
|
was_missing = not self.registry_path.exists()
|
||||||
self.keys = []
|
document, warning = load_json_document(
|
||||||
|
self.registry_path,
|
||||||
|
default={"schema_version": SCHEMA_VERSION, "keys": []},
|
||||||
|
current_version=SCHEMA_VERSION,
|
||||||
|
)
|
||||||
|
self.load_warning = warning
|
||||||
|
self.keys = [KeyEntry.from_dict(k) for k in document.get("keys", [])]
|
||||||
|
|
||||||
|
# Uniformise le format des empreintes : les cles importees avant l'ajout de
|
||||||
|
# compute_fingerprint() portaient un simple fragment du base64 plutot qu'un
|
||||||
|
# vrai condense SHA256. On les corrige silencieusement au chargement.
|
||||||
|
changed = False
|
||||||
|
for entry in self.keys:
|
||||||
|
if not entry.public_key_text:
|
||||||
|
continue
|
||||||
|
correct = compute_fingerprint(entry.public_key_text)
|
||||||
|
if correct and entry.fingerprint != correct:
|
||||||
|
entry.fingerprint = correct
|
||||||
|
changed = True
|
||||||
|
|
||||||
|
if was_missing or changed:
|
||||||
self.save()
|
self.save()
|
||||||
return
|
|
||||||
try:
|
|
||||||
raw = json.loads(self.registry_path.read_text(encoding="utf-8"))
|
|
||||||
except (json.JSONDecodeError, OSError):
|
|
||||||
raw = {"keys": []}
|
|
||||||
self.keys = [KeyEntry.from_dict(k) for k in raw.get("keys", [])]
|
|
||||||
|
|
||||||
def save(self) -> None:
|
def save(self) -> None:
|
||||||
DATA_DIR.mkdir(parents=True, exist_ok=True)
|
payload = {"schema_version": SCHEMA_VERSION, "keys": [k.to_dict() for k in self.keys]}
|
||||||
payload = {"keys": [k.to_dict() for k in self.keys]}
|
save_json_document(self.registry_path, payload)
|
||||||
self.registry_path.write_text(json.dumps(payload, indent=2, ensure_ascii=False), encoding="utf-8")
|
|
||||||
|
|
||||||
def remove(self, key_id: str, delete_files: bool = False) -> None:
|
def remove(self, key_id: str, delete_files: bool = False) -> None:
|
||||||
entry = next((k for k in self.keys if k.id == key_id), None)
|
entry = next((k for k in self.keys if k.id == key_id), None)
|
||||||
@@ -180,12 +217,9 @@ class KeyManager:
|
|||||||
|
|
||||||
priv_path.write_bytes(priv_bytes)
|
priv_path.write_bytes(priv_bytes)
|
||||||
pub_path.write_text(pub_text, encoding="utf-8")
|
pub_path.write_text(pub_text, encoding="utf-8")
|
||||||
try:
|
protect_private_key(priv_path)
|
||||||
priv_path.chmod(0o600)
|
|
||||||
except OSError:
|
|
||||||
pass
|
|
||||||
|
|
||||||
fingerprint = self._fingerprint(pub_bytes)
|
fingerprint = compute_fingerprint(pub_text)
|
||||||
entry = KeyEntry(
|
entry = KeyEntry(
|
||||||
name=name,
|
name=name,
|
||||||
key_type=key_type,
|
key_type=key_type,
|
||||||
@@ -213,13 +247,12 @@ class KeyManager:
|
|||||||
pub_path_str = str(guess)
|
pub_path_str = str(guess)
|
||||||
pub_text = guess.read_text(encoding="utf-8").strip()
|
pub_text = guess.read_text(encoding="utf-8").strip()
|
||||||
|
|
||||||
fingerprint = pub_text.split()[1] if pub_text and len(pub_text.split()) > 1 else ""
|
|
||||||
entry = KeyEntry(
|
entry = KeyEntry(
|
||||||
name=name,
|
name=name,
|
||||||
key_type=key_type,
|
key_type=key_type,
|
||||||
private_path=str(priv_path),
|
private_path=str(priv_path),
|
||||||
public_path=pub_path_str,
|
public_path=pub_path_str,
|
||||||
fingerprint=fingerprint[:16],
|
fingerprint=compute_fingerprint(pub_text) if pub_text else "",
|
||||||
public_key_text=pub_text,
|
public_key_text=pub_text,
|
||||||
)
|
)
|
||||||
self.keys.append(entry)
|
self.keys.append(entry)
|
||||||
@@ -248,15 +281,3 @@ class KeyManager:
|
|||||||
|
|
||||||
def export_private_key(self, entry: KeyEntry, destination: str) -> None:
|
def export_private_key(self, entry: KeyEntry, destination: str) -> None:
|
||||||
shutil.copyfile(entry.private_path, destination)
|
shutil.copyfile(entry.private_path, destination)
|
||||||
|
|
||||||
@staticmethod
|
|
||||||
def _fingerprint(pub_bytes: bytes) -> str:
|
|
||||||
import base64
|
|
||||||
import hashlib
|
|
||||||
|
|
||||||
parts = pub_bytes.split()
|
|
||||||
if len(parts) < 2:
|
|
||||||
return ""
|
|
||||||
raw = base64.b64decode(parts[1])
|
|
||||||
digest = hashlib.sha256(raw).digest()
|
|
||||||
return "SHA256:" + base64.b64encode(digest).decode("utf-8").rstrip("=")
|
|
||||||
|
|||||||
Reference in New Issue
Block a user