Credential persistence for tools and channels.
Stores credentials in ~/.openjarvis/credentials.toml with 0o600 permissions.
Thread-safe writes via lock. Sets os.environ on save for immediate effect.
Functions
load_credentials
load_credentials(path: Path | None = None) -> dict[str, dict[str, str]]
Load credentials from TOML file.
Source code in src/openjarvis/core/credentials.py
| def load_credentials(path: Path | None = None) -> dict[str, dict[str, str]]:
"""Load credentials from TOML file."""
p = Path(path) if path else _DEFAULT_PATH
if not p.exists():
return {}
with open(p, "rb") as f:
return tomllib.load(f)
|
save_credential
save_credential(tool_name: str, key: str, value: str, *, path: Path | None = None) -> None
Save a single credential key, validate, write file, and set os.environ.
Source code in src/openjarvis/core/credentials.py
| def save_credential(
tool_name: str,
key: str,
value: str,
*,
path: Path | None = None,
) -> None:
"""Save a single credential key, validate, write file, and set os.environ."""
allowed = TOOL_CREDENTIALS.get(tool_name, [])
if key not in allowed:
raise ValueError(f"Unknown credential key '{key}' for tool '{tool_name}'")
stripped = value.strip()
if not stripped:
raise ValueError("Credential value must not be empty")
p = Path(path) if path else _DEFAULT_PATH
with _LOCK:
creds = load_credentials(path=p)
if tool_name not in creds:
creds[tool_name] = {}
creds[tool_name][key] = stripped
p.parent.mkdir(parents=True, exist_ok=True)
lines: list[str] = []
for section, kvs in creds.items():
lines.append(f"[{section}]")
for k, v in kvs.items():
lines.append(f'{k} = "{v}"')
lines.append("")
p.write_text("\n".join(lines))
os.chmod(p, 0o600)
os.environ[key] = stripped
|
get_credential_status
get_credential_status(tool_name: str) -> dict[str, bool]
Return {KEY: bool} for each required key indicating if set in env.
Source code in src/openjarvis/core/credentials.py
| def get_credential_status(tool_name: str) -> dict[str, bool]:
"""Return {KEY: bool} for each required key indicating if set in env."""
keys = TOOL_CREDENTIALS.get(tool_name, [])
return {k: bool(os.environ.get(k)) for k in keys}
|
inject_credentials
inject_credentials(path: Path | None = None) -> None
Load credentials.toml and inject into os.environ. Call at server startup.
Source code in src/openjarvis/core/credentials.py
| def inject_credentials(path: Path | None = None) -> None:
"""Load credentials.toml and inject into os.environ. Call at server startup."""
creds = load_credentials(path=path)
for _tool, kvs in creds.items():
for k, v in kvs.items():
if k not in os.environ:
os.environ[k] = v
|
get_tool_credential(tool_name: str, key: str, *, path: Path | None = None) -> str | None
Read a single credential without polluting os.environ.
Falls back to os.environ if the key is not in credentials.toml,
for backward compatibility with Docker env var workflows.
Source code in src/openjarvis/core/credentials.py
| def get_tool_credential(
tool_name: str,
key: str,
*,
path: Path | None = None,
) -> str | None:
"""Read a single credential without polluting ``os.environ``.
Falls back to ``os.environ`` if the key is not in credentials.toml,
for backward compatibility with Docker env var workflows.
"""
creds = load_credentials(path=path)
tool_creds = creds.get(tool_name, {})
value = tool_creds.get(key)
if value is not None:
return value
return os.environ.get(key) or None
|