Atlas
eb5b67dee2
Product layer on top of the pure engine: a wizard to connect the apps that feed the brain, an encrypted credential vault (Fernet/PBKDF2; app logins never enter brain.db), and 3 working connectors. Connector data flows through the same credential-scrub path as the engine. Atlas Corporation proprietary.
51 lines
1.9 KiB
Python
51 lines
1.9 KiB
Python
# SecondBrain — (c) 2026 Atlas Corporation. All Rights Reserved. Proprietary; see LICENSE.
|
|
"""Encrypted credential vault for connectors.
|
|
|
|
App logins (IMAP passwords, WebDAV app-passwords, OAuth tokens) are stored here,
|
|
encrypted at rest with a master passphrase (Fernet, PBKDF2-HMAC-SHA256, 390k iters).
|
|
brain.db NEVER sees these — connectors read creds from the vault, pull data, and only
|
|
scrubbed text reaches the brain. Set BRAIN_VAULT_PASS to run non-interactively.
|
|
"""
|
|
import os, json, base64, getpass
|
|
from cryptography.fernet import Fernet
|
|
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
|
|
from cryptography.hazmat.primitives import hashes
|
|
|
|
_HOME = os.environ.get("BRAIN_HOME", os.path.expanduser("~/.secondbrain"))
|
|
VAULT = os.path.join(_HOME, "vault.enc")
|
|
SALT_F = os.path.join(_HOME, "vault.salt")
|
|
|
|
def _salt():
|
|
os.makedirs(_HOME, exist_ok=True)
|
|
if not os.path.exists(SALT_F):
|
|
s = os.urandom(16)
|
|
with open(SALT_F, "wb") as f:
|
|
f.write(s)
|
|
os.chmod(SALT_F, 0o600)
|
|
return s
|
|
with open(SALT_F, "rb") as f:
|
|
return f.read()
|
|
|
|
def _key(passphrase, salt):
|
|
kdf = PBKDF2HMAC(algorithm=hashes.SHA256(), length=32, salt=salt, iterations=390000)
|
|
return base64.urlsafe_b64encode(kdf.derive(passphrase.encode()))
|
|
|
|
def _pass(prompt):
|
|
return os.environ.get("BRAIN_VAULT_PASS") or getpass.getpass(prompt)
|
|
|
|
def load(passphrase=None):
|
|
if not os.path.exists(VAULT):
|
|
return {}
|
|
passphrase = passphrase or _pass("Vault passphrase: ")
|
|
f = Fernet(_key(passphrase, _salt()))
|
|
with open(VAULT, "rb") as fh:
|
|
return json.loads(f.decrypt(fh.read()).decode())
|
|
|
|
def save(data, passphrase=None):
|
|
passphrase = passphrase or _pass("Vault passphrase (set/confirm): ")
|
|
os.makedirs(_HOME, exist_ok=True)
|
|
f = Fernet(_key(passphrase, _salt()))
|
|
tok = f.encrypt(json.dumps(data).encode())
|
|
with open(VAULT, "wb") as fh:
|
|
fh.write(tok)
|
|
os.chmod(VAULT, 0o600)
|