144 lines
4.6 KiB
Python
144 lines
4.6 KiB
Python
# -*- coding: utf-8 -*-
|
|
import pytest
|
|
from cryptography.fernet import Fernet
|
|
|
|
from db.accounts_repo import insert_credential
|
|
|
|
|
|
class TestCredentialsRepo:
|
|
def test_encrypted_insert_and_read(self, monkeypatch, clean_db):
|
|
from db.credentials_repo import (
|
|
insert_credential_encrypted,
|
|
list_credentials_by_account,
|
|
read_credential_plaintext,
|
|
)
|
|
from service.master_key import ENV_MASTER_KEY
|
|
|
|
key = Fernet.generate_key()
|
|
monkeypatch.setenv(ENV_MASTER_KEY, key.decode("ascii"))
|
|
|
|
conn = clean_db
|
|
now = 1_700_000_000
|
|
aid = insert_credential_encrypted(
|
|
conn,
|
|
account_id=7,
|
|
platform_key="icbc_sim",
|
|
credential_label="工行密码",
|
|
credential_type="password",
|
|
plaintext="my-password",
|
|
expires_at=None,
|
|
extra_json="{}",
|
|
now=now,
|
|
)
|
|
conn.commit()
|
|
pt = read_credential_plaintext(conn, aid)
|
|
assert pt == "my-password"
|
|
assert "my-password" not in "".join(
|
|
str(v) for v in list_credentials_by_account(conn, 7)
|
|
)
|
|
|
|
def test_insert_validation(self, monkeypatch, clean_db):
|
|
from db.credentials_repo import insert_credential_encrypted, insert_credential_external
|
|
from service.master_key import ENV_MASTER_KEY
|
|
|
|
monkeypatch.setenv(ENV_MASTER_KEY, Fernet.generate_key().decode("ascii"))
|
|
conn = clean_db
|
|
now = 1
|
|
with pytest.raises(ValueError, match="credential_type"):
|
|
insert_credential_encrypted(
|
|
conn,
|
|
account_id=None,
|
|
platform_key="x",
|
|
credential_label="l",
|
|
credential_type="weird",
|
|
plaintext="x",
|
|
expires_at=None,
|
|
extra_json="{}",
|
|
now=now,
|
|
)
|
|
with pytest.raises(ValueError, match="plaintext"):
|
|
insert_credential_encrypted(
|
|
conn,
|
|
account_id=None,
|
|
platform_key="x",
|
|
credential_label="l",
|
|
credential_type="password",
|
|
plaintext="",
|
|
expires_at=None,
|
|
extra_json="{}",
|
|
now=now,
|
|
)
|
|
with pytest.raises(ValueError, match="secret_storage"):
|
|
insert_credential_external(
|
|
conn,
|
|
account_id=None,
|
|
platform_key="x",
|
|
credential_label="l",
|
|
credential_type="password",
|
|
secret_storage="bogus",
|
|
secret_ref=None,
|
|
secret_mask="",
|
|
expires_at=None,
|
|
extra_json="{}",
|
|
now=now,
|
|
)
|
|
|
|
def test_list_has_no_sensitive_keys(self, monkeypatch, clean_db):
|
|
from db.credentials_repo import insert_credential_encrypted, list_credentials_by_account
|
|
from service.master_key import ENV_MASTER_KEY
|
|
|
|
monkeypatch.setenv(ENV_MASTER_KEY, Fernet.generate_key().decode("ascii"))
|
|
conn = clean_db
|
|
now = 2
|
|
insert_credential_encrypted(
|
|
conn,
|
|
account_id=9,
|
|
platform_key="p",
|
|
credential_label="a",
|
|
credential_type="api_key",
|
|
plaintext="k1",
|
|
expires_at=None,
|
|
extra_json="{}",
|
|
now=now,
|
|
)
|
|
insert_credential_encrypted(
|
|
conn,
|
|
account_id=9,
|
|
platform_key="p",
|
|
credential_label="b",
|
|
credential_type="api_token",
|
|
plaintext="k2",
|
|
expires_at=None,
|
|
extra_json="{}",
|
|
now=now + 1,
|
|
)
|
|
conn.commit()
|
|
rows = list_credentials_by_account(conn, 9)
|
|
assert len(rows) == 2
|
|
for d in rows:
|
|
assert "plaintext" not in d
|
|
assert "secret_ciphertext" not in d
|
|
|
|
def test_none_storage_read_fails(self, clean_db):
|
|
from db.credentials_repo import CredentialNotEncryptedError, read_credential_plaintext
|
|
|
|
conn = clean_db
|
|
now = 3
|
|
cid = insert_credential(
|
|
conn,
|
|
account_id=None,
|
|
platform_key="p",
|
|
credential_label="n",
|
|
credential_type="note",
|
|
secret_storage="none",
|
|
secret_ref=None,
|
|
secret_mask="",
|
|
expires_at=None,
|
|
extra_json="{}",
|
|
now=now,
|
|
)
|
|
conn.commit()
|
|
with pytest.raises(CredentialNotEncryptedError) as ei:
|
|
read_credential_plaintext(conn, cid)
|
|
assert ei.value.code == "ERR_CREDENTIAL_NOT_ENCRYPTED"
|