feat: add secure secret provider
This commit is contained in:
@@ -3,6 +3,7 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import getpass
|
||||
import sys
|
||||
from collections.abc import Callable, Sequence
|
||||
|
||||
@@ -16,6 +17,7 @@ from javis.memory.sqlite_store import (
|
||||
)
|
||||
from javis.providers.base import ProviderError
|
||||
from javis.providers.ollama import OllamaProvider
|
||||
from javis.security.secrets import SecretProvider, SecretStoreError
|
||||
|
||||
InputFunction = Callable[[str], str]
|
||||
OutputFunction = Callable[[str], None]
|
||||
@@ -34,9 +36,49 @@ def build_parser() -> argparse.ArgumentParser:
|
||||
subparsers = parser.add_subparsers(dest="command", required=True)
|
||||
chat_parser = subparsers.add_parser("chat", help="lokalen Textchat starten")
|
||||
chat_parser.add_argument("--session", help="vorhandene Sitzungs-ID laden")
|
||||
secrets_parser = subparsers.add_parser("secrets", help="lokale Geheimwerte sicher verwalten")
|
||||
secret_subparsers = secrets_parser.add_subparsers(dest="secret_action", required=True)
|
||||
set_parser = secret_subparsers.add_parser("set", help="Geheimwert verdeckt setzen")
|
||||
set_parser.add_argument("secret_name", choices=["gemini"])
|
||||
secret_subparsers.add_parser("status", help="nur Verfügbarkeit anzeigen")
|
||||
delete_parser = secret_subparsers.add_parser("delete", help="Geheimwert löschen")
|
||||
delete_parser.add_argument("secret_name", choices=["gemini"])
|
||||
return parser
|
||||
|
||||
|
||||
def run_secrets(
|
||||
action: str,
|
||||
*,
|
||||
provider: SecretProvider | None = None,
|
||||
password_fn: Callable[[str], str] = getpass.getpass,
|
||||
output: OutputFunction = print,
|
||||
) -> int:
|
||||
secret_provider = provider or SecretProvider()
|
||||
try:
|
||||
if action == "status":
|
||||
state = "verfügbar" if secret_provider.gemini_key_available() else "nicht verfügbar"
|
||||
output(f"Gemini-Schlüssel: {state}")
|
||||
return 0
|
||||
if action == "set":
|
||||
value = password_fn("Gemini API-Key: ")
|
||||
secret_provider.set_gemini_key(value)
|
||||
output("Gemini-Schlüssel wurde sicher gespeichert.")
|
||||
return 0
|
||||
if action == "delete":
|
||||
deleted = secret_provider.delete_gemini_key()
|
||||
output(
|
||||
"Gemini-Schlüssel wurde gelöscht."
|
||||
if deleted
|
||||
else "Gemini-Schlüssel war nicht verfügbar."
|
||||
)
|
||||
return 0
|
||||
except SecretStoreError as exc:
|
||||
output(f"Fehler: {exc}")
|
||||
return 2
|
||||
output("Unbekannte Secret-Aktion.")
|
||||
return 2
|
||||
|
||||
|
||||
def _show_history(service: ChatService, session_id: str, output: OutputFunction) -> None:
|
||||
loaded = service.load_session(session_id)
|
||||
if not loaded.messages:
|
||||
@@ -155,6 +197,12 @@ def main(
|
||||
output: OutputFunction = print,
|
||||
) -> int:
|
||||
args = build_parser().parse_args(argv)
|
||||
if args.command == "secrets":
|
||||
return run_secrets(
|
||||
args.secret_action,
|
||||
password_fn=input_fn if input_fn is not input else getpass.getpass,
|
||||
output=output,
|
||||
)
|
||||
if args.command != "chat":
|
||||
return 2
|
||||
|
||||
|
||||
@@ -0,0 +1,74 @@
|
||||
"""Secret storage backed by the operating system credential store."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Protocol, runtime_checkable
|
||||
|
||||
import keyring
|
||||
from keyring.errors import KeyringError, PasswordDeleteError
|
||||
|
||||
SERVICE_NAME = "javis-ai.local"
|
||||
GEMINI_ACCOUNT = "gemini-api-key"
|
||||
|
||||
|
||||
class SecretStoreError(RuntimeError):
|
||||
"""A secret operation failed without exposing the secret value."""
|
||||
|
||||
|
||||
@runtime_checkable
|
||||
class SecretBackend(Protocol):
|
||||
def get_password(self, service: str, account: str) -> str | None: ...
|
||||
|
||||
def set_password(self, service: str, account: str, value: str) -> None: ...
|
||||
|
||||
def delete_password(self, service: str, account: str) -> None: ...
|
||||
|
||||
|
||||
class KeyringBackend:
|
||||
"""Adapter around Python keyring and the active OS credential backend."""
|
||||
|
||||
def get_password(self, service: str, account: str) -> str | None:
|
||||
return keyring.get_password(service, account)
|
||||
|
||||
def set_password(self, service: str, account: str, value: str) -> None:
|
||||
keyring.set_password(service, account, value)
|
||||
|
||||
def delete_password(self, service: str, account: str) -> None:
|
||||
keyring.delete_password(service, account)
|
||||
|
||||
|
||||
class SecretProvider:
|
||||
"""Named Javis secrets without any value-bearing status output."""
|
||||
|
||||
def __init__(self, backend: SecretBackend | None = None) -> None:
|
||||
self._backend = backend or KeyringBackend()
|
||||
|
||||
def get_gemini_key(self) -> str | None:
|
||||
try:
|
||||
return self._backend.get_password(SERVICE_NAME, GEMINI_ACCOUNT)
|
||||
except KeyringError as exc:
|
||||
raise SecretStoreError(
|
||||
"Der Betriebssystem-Schlüsselspeicher ist nicht verfügbar."
|
||||
) from exc
|
||||
|
||||
def gemini_key_available(self) -> bool:
|
||||
return bool(self.get_gemini_key())
|
||||
|
||||
def set_gemini_key(self, value: str) -> None:
|
||||
if not value or not value.strip():
|
||||
raise SecretStoreError("Ein leerer Gemini-Schlüssel wird nicht gespeichert.")
|
||||
try:
|
||||
self._backend.set_password(SERVICE_NAME, GEMINI_ACCOUNT, value.strip())
|
||||
except KeyringError as exc:
|
||||
raise SecretStoreError(
|
||||
"Der Gemini-Schlüssel konnte nicht sicher gespeichert werden."
|
||||
) from exc
|
||||
|
||||
def delete_gemini_key(self) -> bool:
|
||||
try:
|
||||
self._backend.delete_password(SERVICE_NAME, GEMINI_ACCOUNT)
|
||||
except PasswordDeleteError:
|
||||
return False
|
||||
except KeyringError as exc:
|
||||
raise SecretStoreError("Der Gemini-Schlüssel konnte nicht gelöscht werden.") from exc
|
||||
return True
|
||||
Reference in New Issue
Block a user