feat(security): Implement KMS encryption for credentials - #1117
feat(security): Implement KMS encryption for credentials#1117vprashrex wants to merge 4 commits into
Conversation
… update related tests
📝 WalkthroughWalkthroughKMS-backed credentials now use AES-GCM envelope encryption with ChangesKMS envelope encryption
Estimated code review effort: 3 (Moderate) | ~25 minutes Sequence Diagram(s)sequenceDiagram
participant CredentialService
participant AWSKMS
participant AESGCM
CredentialService->>AWSKMS: GenerateDataKey
AWSKMS-->>CredentialService: Return wrapped and plaintext data keys
CredentialService->>AESGCM: Encrypt credentials locally
AESGCM-->>CredentialService: Return nonce and ciphertext
CredentialService->>CredentialService: Store kms.v2 envelope
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
github-actions
Bot
changed the title
OpenAPI changes ⚪ No API surface changesNote This PR does not modify the API contract.
|
Codecov Report❌ Patch coverage is
📢 Thoughts on this report? Let us know! |
| def upgrade(): | ||
| bind = op.get_bind() | ||
| with Session(bind=bind) as session: | ||
| execute_credential_reencrypt(session=session) |
There was a problem hiding this comment.
@kartpop This will invoke the sync function and execute execute_credential_reencrypt
There was a problem hiding this comment.
there are only 497 credential rows in prod db. So sync option should be ok.
There was a problem hiding this comment.
bdw should we do that inside a migration or a separate private endpoint @vprashrex
There was a problem hiding this comment.
Yeah, we discussed this before. Invoking it via an endpoint isn’t the right approach because it should be idempotent and run only once. A migration is the better choice since it guarantees a one-time execution and keeps the change consistent across environments.
| """ | ||
| try: | ||
| if encrypted_credentials.startswith(KMS_CIPHERTEXT_PREFIX): | ||
| if encrypted_credentials.startswith(KMS_ENVELOPE_PREFIX): |
There was a problem hiding this comment.
This if-elif condition checks the format of the existing credential and then decrypts it using the corresponding decryption logic for that format.
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@backend/app/tests/core/test_security.py`:
- Around line 84-140: Update the new test methods in
backend/app/tests/core/test_security.py:84-140 and the test at
backend/app/tests/services/credentials/test_reencrypt.py:43 to annotate each
fixture parameter with its concrete fixture type and add -> None return
annotations. The site
backend/app/alembic/versions/076_reencrypt_credentials_envelope.py:25-33
requires no direct change.
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro Plus
Run ID: 08063f7c-75f6-4688-a03d-f01dd8ea6c26
📒 Files selected for processing (5)
backend/app/alembic/versions/076_reencrypt_credentials_envelope.pybackend/app/core/security.pybackend/app/tests/core/test_security.pybackend/app/tests/services/credentials/test_reencrypt.pydocs/wiki/modules/platform.md
| def test_kms_v2_envelope_roundtrip(self, kms_key): | ||
| creds = {"openai": {"api_key": "sk-envelope-123"}} | ||
|
|
||
| encrypted = encrypt_credentials(creds) | ||
|
|
||
| assert encrypted.startswith(KMS_ENVELOPE_PREFIX) | ||
| segments = encrypted[len(KMS_ENVELOPE_PREFIX) :].split(":") | ||
| assert len(segments) == 3 | ||
| for seg in segments: | ||
| base64.b64decode(seg) # each segment must be valid base64 | ||
| assert decrypt_credentials(encrypted) == creds | ||
|
|
||
| def test_kms_v2_large_payload_over_4096_bytes(self, kms_key): | ||
| # Direct KMS encrypt caps at 4096 bytes; envelope encryption has no such limit. | ||
| creds = { | ||
| "service_account": {"private_key": "k" * 5000, "client_email": "svc@x"} | ||
| } | ||
| assert len(json.dumps(creds).encode()) > 4096 | ||
|
|
||
| encrypted = encrypt_credentials(creds) | ||
|
|
||
| assert encrypted.startswith(KMS_ENVELOPE_PREFIX) | ||
| assert decrypt_credentials(encrypted) == creds | ||
|
|
||
| def test_v1_row_still_decrypts_with_v2_active(self, kms_key): | ||
| creds = {"api_key": "sk-v1-legacy"} | ||
| blob = security._kms_client.encrypt( | ||
| KeyId=kms_key, Plaintext=json.dumps(creds).encode() | ||
| )["CiphertextBlob"] | ||
| v1_ciphertext = KMS_CIPHERTEXT_PREFIX + base64.b64encode(blob).decode() | ||
|
|
||
| assert decrypt_credentials(v1_ciphertext) == creds | ||
|
|
||
| def test_new_writes_produce_v2(self, kms_key): | ||
| encrypted = encrypt_credentials({"api_key": "sk-new"}) | ||
|
|
||
| assert encrypted.startswith(KMS_ENVELOPE_PREFIX) | ||
| assert not encrypted.startswith(KMS_CIPHERTEXT_PREFIX) | ||
|
|
||
| def test_v2_tampered_ciphertext_raises(self, kms_key): | ||
| encrypted = encrypt_credentials({"api_key": "sk-tamper"}) | ||
| wrapped_b64, nonce_b64, ct_b64 = encrypted[len(KMS_ENVELOPE_PREFIX) :].split( | ||
| ":" | ||
| ) | ||
| ct = bytearray(base64.b64decode(ct_b64)) | ||
| ct[0] ^= 0xFF | ||
| tampered = ( | ||
| f"{KMS_ENVELOPE_PREFIX}{wrapped_b64}:{nonce_b64}:" | ||
| f"{base64.b64encode(bytes(ct)).decode()}" | ||
| ) | ||
|
|
||
| with pytest.raises(ValueError, match="Failed to decrypt credentials"): | ||
| decrypt_credentials(tampered) | ||
|
|
||
| def test_v2_wrong_segment_count_raises(self, kms_key): | ||
| with pytest.raises(ValueError, match="Failed to decrypt credentials"): | ||
| decrypt_credentials(f"{KMS_ENVELOPE_PREFIX}onlyoneseg") |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== Candidate files =="
git ls-files | rg 'backend/app/(tests/core/test_security.py|alembic/versions/076_reencrypt_credentials_envelope.py|tests/services/credentials/test_reencrypt.py)$' || true
echo
echo "== test_security.py relevant section =="
if [ -f backend/app/tests/core/test_security.py ]; then
nl -ba backend/app/tests/core/test_security.py | sed -n '75,145p'
fi
echo
echo "== alembic migration relevant section =="
if [ -f backend/app/alembic/versions/076_reencrypt_credentials_envelope.py ]; then
nl -ba backend/app/alembic/versions/076_reencrypt_credentials_envelope.py | sed -n '1,60p'
fi
echo
echo "== test_reencrypt.py relevant section =="
if [ -f backend/app/tests/services/credentials/test_reencrypt.py ]; then
nl -ba backend/app/tests/services/credentials/test_reencrypt.py | sed -n '30,55p'
fi
echo
echo "== Existing typing fixtures in tests/security/alembic to infer exact fixture names =="
rg -n 'def (kms_key|monkeypatch)|class|Session|encrypt_credentials|decrypt_credentials|KMS_ENVELOPE_PREFIX' backend/app/tests/core/test_security.py backend/app/alembic/versions/076_reencrypt_credentials_envelope.py backend/app/tests/services/credentials/test_reencrypt.py | sed -n '1,120p'Repository: ProjectTech4DevAI/kaapi-backend
Length of output: 437
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== Candidate files =="
git ls-files | rg 'backend/app/(tests/core/test_security.py|alembic/versions/076_reencrypt_credentials_envelope.py|tests/services/credentials/test_reencrypt.py)$' || true
echo
echo "== test_security.py relevant section =="
if [ -f backend/app/tests/core/test_security.py ]; then
nl -ba backend/app/tests/core/test_security.py | sed -n '75,145p'
fi
echo
echo "== alembic migration relevant section =="
if [ -f backend/app/alembic/versions/076_reencrypt_credentials_envelope.py ]; then
nl -ba backend/app/alembic/versions/076_reencrypt_credentials_envelope.py | sed -n '1,60p'
fi
echo
echo "== test_reencrypt.py relevant section =="
if [ -f backend/app/tests/services/credentials/test_reencrypt.py ]; then
nl -ba backend/app/tests/services/credentials/test_reencrypt.py | sed -n '30,55p'
fi
echo
echo "== Existing typing fixtures in tests/security/alembic to infer exact fixture names =="
rg -n 'def (kms_key|monkeypatch)|class|Session|encrypt_credentials|decrypt_credentials|KMS_ENVELOPE_PREFIX' backend/app/tests/core/test_security.py backend/app/alembic/versions/076_reencrypt_credentials_envelope.py backend/app/tests/services/credentials/test_reencrypt.py | sed -n '1,120p'Repository: ProjectTech4DevAI/kaapi-backend
Length of output: 437
🏁 Script executed:
#!/bin/bash
set -euo pipefail
python3 - <<'PY'
from pathlib import Path
files = [
Path("backend/app/tests/core/test_security.py"),
Path("backend/app/alembic/versions/076_reencrypt_credentials_envelope.py"),
Path("backend/app/tests/services/credentials/test_reencrypt.py"),
]
for path in files:
print(f"\n== {path} lines 1-180 or relevant ==")
lines = path.read_text(encoding="utf-8").splitlines()
start, end = max(1, min(180, len(lines))), len(lines)
if path.name == "backend/app/tests/core/test_security.py":
start, end = max(1, 75), min(145, len(lines))
if path.name == "backend/app/alembic/versions/076_reencrypt_credentials_envelope.py":
start, end = max(1, 1), min(60, len(lines))
if path.name == "backend/app/tests/services/credentials/test_reencrypt.py":
start, end = max(1, 30), min(55, len(lines))
for i in range(start, end + 1):
print(f"{i:5}: {lines[i-1]}")
PYRepository: ProjectTech4DevAI/kaapi-backend
Length of output: 9435
🏁 Script executed:
#!/bin/bash
set -euo pipefail
python3 - <<'PY'
from pathlib import Path
for path in [
Path("backend/app/tests/core/test_security.py"),
Path("backend/app/alembic/versions/076_reencrypt_credentials_envelope.py"),
Path("backend/app/tests/services/credentials/test_reencrypt.py"),
]:
print(f"\n== {path} lines 1-120 ==")
lines = path.read_text(encoding="utf-8").splitlines()
start, end = max(1, min(120, len(lines))), len(lines)
for i in range(start, end + 1):
print(f"{i:5}: {lines[i-1]}")
PY
echo
echo "== Read-only annotation extractor for target lines =="
python3 - <<'PY'
from pathlib import Path
import re
files = [
Path("backend/app/tests/core/test_security.py"),
Path("backend/app/alembic/versions/076_reencrypt_credentials_envelope.py"),
Path("backend/app/tests/services/credentials/test_reencrypt.py"),
]
def target_lines(path: Path):
text = path.read_text()
all_lines = text.splitlines()
matches = list(re.finditer(r'(def|async def)\s+([A-Za-z_]\w*)', text))
results = []
for m in matches:
lineno = text.count("\n", 0, m.start()) + 1
if (
path.name == "backend/app/tests/core/test_security.py" and lineno in range(84, 140 + 1)
) or (
path.name == "backend/app/alembic/versions/076_reencrypt_credentials_envelope.py" and lineno in range(25, 33 + 1)
) or (
path.name == "backend/app/tests/services/credentials/test_reencrypt.py" and lineno == 43
):
line = all_lines[lineno - 1]
has_self = bool(re.search(r'\bself\b', line))
has_self_annot = bool(re.search(r'\bself\s*:\s*[^=,\n:]+', line))
has_return_arrow = " -> " in line
results.append((lineno, line, has_self, has_self_annot, has_return_arrow))
return results
for path in files:
print(f"{path}")
for item in target_lines(path):
print(f" line {item[0]} self_annot={item[3]} return_annot={item[4]}")
print(f" {item[1]}")
PYRepository: ProjectTech4DevAI/kaapi-backend
Length of output: 12407
Add narrow type hints to the new KMS credential tests.
The test method signatures in backend/app/tests/core/test_security.py lines 84-140 still omit concrete annotations. Add each fixture parameter type and -> None; apply the same rule to the new test method at backend/app/tests/services/credentials/test_reencrypt.py#43.
🧰 Tools
🪛 ast-grep (0.45.0)
[info] 100-100: use jsonify instead of json.dumps for JSON output
Context: json.dumps(creds)
Note: [CWE-116] Improper Encoding or Escaping of Output.
(use-jsonify)
[info] 110-110: use jsonify instead of json.dumps for JSON output
Context: json.dumps(creds)
Note: [CWE-116] Improper Encoding or Escaping of Output.
(use-jsonify)
📍 Affects 3 files
backend/app/tests/core/test_security.py#L84-L140(this comment)backend/app/alembic/versions/076_reencrypt_credentials_envelope.py#L25-L33backend/app/tests/services/credentials/test_reencrypt.py#L43-L43
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@backend/app/tests/core/test_security.py` around lines 84 - 140, Update the
new test methods in backend/app/tests/core/test_security.py:84-140 and the test
at backend/app/tests/services/credentials/test_reencrypt.py:43 to annotate each
fixture parameter with its concrete fixture type and add -> None return
annotations. The site
backend/app/alembic/versions/076_reencrypt_credentials_envelope.py:25-33
requires no direct change.
Source: Coding guidelines
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@backend/app/services/credentials/reencrypt.py`:
- Around line 47-49: Update the row handling in the re-encryption migration to
validate the complete v2 envelope shape, including its three required base64
segments, before skipping it; do not rely on startswith(KMS_ENVELOPE_PREFIX)
alone. Malformed prefixed values must continue to decrypt_credentials so they
raise and roll back the migration. Revise the nearby comment to explain that
skipping a valid v2 envelope keeps the backfill idempotent.
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro Plus
Run ID: 4378ef1d-cc09-4b9f-8020-b851b011db78
📒 Files selected for processing (1)
backend/app/services/credentials/reencrypt.py
kartpop
left a comment
kartpop
left a comment
There was a problem hiding this comment.
let me know if you want to discuss the review comments
| """Re-encrypt every stored credential through the current encryption scheme. | ||
|
|
||
| Atomic: one transaction, all-or-nothing — any failure rolls back and re-raises. | ||
| Sole caller is the one-shot Alembic migration 073; no route/CRUD/Celery path. |
| converted = 0 | ||
| try: | ||
| for row in rows: | ||
| if row.credential.startswith(KMS_ENVELOPE_PREFIX): |
There was a problem hiding this comment.
actually, will this mess up roll back by any chance? should we make in place changes to kms.v1 migration?
also later on if we have kms.v3 or something (or if we choose to rotate the KMS CMK), then we have to be careful not to re-use this function and instead write a new one - otherwise it will silently skip the entries we would want it to reencrypt
should we instead have two different versions of the _reencrypt function? there would be some code duplication, but will probably keep things cleaner - less chances of error
There was a problem hiding this comment.
Like we did for base64 to S3 uri for TTS and STT payload, we had added a batchwise checkpointing (roll back txn if error occurs per 20 rows, stop further execution, restart). This enabled the system to have a limited blast radius.
|
yeah i also thought same regarding this. so i created a function named
reencrypt_fernet which will rencrypt credential to fernet on downgrade
…On Mon, 10 Aug 2026 at 4:39 PM, Kartikeya Pophali ***@***.***> wrote:
***@***.**** requested changes on this pull request.
let me know if you want to discuss the review comments
------------------------------
In backend/app/services/credentials/reencrypt.py
<#1117 (comment)>
:
>
now even 076 is calling
------------------------------
In backend/app/services/credentials/reencrypt.py
<#1117 (comment)>
:
> @@ -39,6 +44,9 @@ def _reencrypt(session: Session) -> dict[str, int]:
converted = 0
try:
for row in rows:
+ if row.credential.startswith(KMS_ENVELOPE_PREFIX):
actually, will this mess up roll back by any chance? should we make in
place changes to kms.v1 migration?
also later on if we have kms.v3 or something (or if we choose to rotate
the KMS CMK), then we have to be careful not to re-use this function and
instead write a new one - otherwise it will silently skip the entries we
would want it to reencrypt
should we instead have two different versions of the _reencrypt function?
there would be some code duplication, but will probably keep things cleaner
- less chances of error
—
Reply to this email directly, view it on GitHub
<#1117?email_source=notifications&email_token=ARCURUP22UC6EG3EINQXKJD5JGUQNA5CNFSNUABKM5UWIORPF5TWS5BNNB2WEL2QOVWGYUTFOF2WK43UKJSXM2LFO4XTIOBZGYYDOMBVGU42M4TFMFZW63VGMFZXG2LHN2SWK5TFNZ2KYZTPN52GK4S7MNWGSY3L#pullrequestreview-4896070559>,
or unsubscribe
<https://github.com/notifications/unsubscribe-auth/ARCURUMHPTHHFD5OMJ4SDVL5JGUQNAVCNFSNUABFKJSXA33TNF2G64TZHM4TINBTGQ4DGOJTHNEXG43VMU5TKMBYGYZTMOJRGE4KC5QC>
.
You are receiving this because you were assigned.Message ID:
***@***.***>
|
…ntials and update re-encryption logic
There was a problem hiding this comment.
Actionable comments posted: 2
🧹 Nitpick comments (2)
backend/app/services/credentials/reencrypt.py (1)
113-117: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winExplain the preservation guarantee in the inline comment.
The comment restates the prefix condition. State why non-KMS rows are skipped: the downgrade must preserve Fernet rows to remain idempotent and avoid unnecessary token replacement.
Proposed change
- # Convert only KMS rows; anything without a KMS prefix is already Fernet. + # Preserve Fernet rows so this downgrade is idempotent and avoids token churn.As per coding guidelines, “Comments must explain why rather than what; avoid comments that restate code or narrate self-evident operations.”
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@backend/app/services/credentials/reencrypt.py` around lines 113 - 117, Update the inline comment above the prefix check to explain that non-KMS rows are skipped to preserve existing Fernet credentials, keeping the downgrade idempotent and avoiding unnecessary token replacement.Source: Coding guidelines
backend/app/core/security.py (1)
214-221: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winName the envelope protocol parameters.
Line 216 and Line 220 embed the KMS key specification and AES-GCM nonce length. Define named constants near the envelope prefix constants. This makes protocol changes explicit and auditable.
Proposed change
+KMS_DATA_KEY_SPEC = "AES_256" +AES_GCM_NONCE_BYTES = 12 + - KeySpec="AES_256", + KeySpec=KMS_DATA_KEY_SPEC, ... - nonce = os.urandom(12) + nonce = os.urandom(AES_GCM_NONCE_BYTES)As per coding guidelines, “Do not use magic values; extract repeated literals into constants, enums, or settings.”
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@backend/app/core/security.py` around lines 214 - 221, Define named constants near the existing envelope prefix constants for the KMS key specification and AES-GCM nonce length, then update the encryption flow around get_kms_client().generate_data_key and os.urandom() to use those constants instead of the embedded literals. Preserve the current AES-256 and 12-byte nonce protocol values.Source: Coding guidelines
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@backend/app/alembic/versions/076_reencrypt_credentials_envelope.py`:
- Around line 33-37: Add explicit -> None return annotations to both migration
functions, upgrade and downgrade, while leaving their existing migration logic
unchanged.
In `@backend/app/core/security.py`:
- Line 235: Define a recursive JSON value type alias in security.py and replace
dict[str, Any] with the corresponding JSON-object type in both encrypt_fernet
and decrypt_fernet signatures. Ensure the annotations permit only
JSON-serializable primitive, array, and object values so invalid credentials are
rejected at the crypto contract boundary.
---
Nitpick comments:
In `@backend/app/core/security.py`:
- Around line 214-221: Define named constants near the existing envelope prefix
constants for the KMS key specification and AES-GCM nonce length, then update
the encryption flow around get_kms_client().generate_data_key and os.urandom()
to use those constants instead of the embedded literals. Preserve the current
AES-256 and 12-byte nonce protocol values.
In `@backend/app/services/credentials/reencrypt.py`:
- Around line 113-117: Update the inline comment above the prefix check to
explain that non-KMS rows are skipped to preserve existing Fernet credentials,
keeping the downgrade idempotent and avoiding unnecessary token replacement.
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro Plus
Run ID: da8f3221-e90a-485b-8679-db3a4adf14e8
📒 Files selected for processing (3)
backend/app/alembic/versions/076_reencrypt_credentials_envelope.pybackend/app/core/security.pybackend/app/services/credentials/reencrypt.py
| def downgrade(): | ||
| # Reverse the envelope backfill: convert kms.v1/kms.v2 rows back to legacy Fernet. | ||
| bind = op.get_bind() | ||
| with Session(bind=bind) as session: | ||
| execute_credential_reencrypt_fernet(session=session) |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
ast-grep outline backend/app/alembic/versions/076_reencrypt_credentials_envelope.py --items all --type function
rg -n '^def (upgrade|downgrade)\(' backend/app/alembic/versions/076_reencrypt_credentials_envelope.pyRepository: ProjectTech4DevAI/kaapi-backend
Length of output: 314
🏁 Script executed:
#!/bin/bash
set -euo pipefail
sed -n '1,90c' <<'EOF'
import os
from pathlib import Path
from typing import Any
from pydantic import BaseModel, ValidationError
class SimpleConfig(BaseModel):
value: str
def validate_value(self) -> str:
if not self.value:
raise ValidationError("value must exist")
return self.value
def process_config(config: SimpleConfig) -> str:
return config.validate_value()
if __name__ == "__main__":
config = SimpleConfig(value="example")
print(process_config(config))
EOFRepository: ProjectTech4DevAI/kaapi-backend
Length of output: 233
🏁 Script executed:
#!/bin/bash
set -euo pipefail
sed -n '1,90p' backend/app/alembic/versions/076_reencrypt_credentials_envelope.py
python3 - <<'PY'
import ast
from pathlib import Path
path = Path("backend/app/alembic/versions/076_reencrypt_credentials_envelope.py")
tree = ast.parse(path.read_text(), filename=str(path))
for node in ast.walk(tree):
if isinstance(node, ast.FunctionDef) and node.name in ("upgrade", "downgrade"):
print(node.name, "returns_annotation=", ast.unparse(node.returns) if node.returns else None)
PYRepository: ProjectTech4DevAI/kaapi-backend
Length of output: 1346
Add return annotations to upgrade and downgrade.
Both migration functions return None, so declare -> None on each function definition.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@backend/app/alembic/versions/076_reencrypt_credentials_envelope.py` around
lines 33 - 37, Add explicit -> None return annotations to both migration
functions, upgrade and downgrade, while leaving their existing migration logic
unchanged.
Source: Coding guidelines
| raise ValueError("Failed to encrypt credentials") | ||
|
|
||
|
|
||
| def encrypt_fernet(credentials: dict[str, Any]) -> str: |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
ast-grep outline backend/app/core/security.py --items all --type function --match 'encrypt_credentials|encrypt_fernet|decrypt_credentials'
rg -n -C 2 'TypeAlias|JSONValue|JsonValue|dict\[str, Any\]' backend/appRepository: ProjectTech4DevAI/kaapi-backend
Length of output: 50388
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== security.py definitions =="
sed -n '180,260p' backend/app/core/security.py
echo
echo "== imports and Any usage in core/security.py =="
sed -n '1,40p' backend/app/core/security.py
rg -n '^(import|from|def |@)' backend/app/core/security.py
echo
echo "== search reusable aliases excluding any backend/app =="
rg -n -C 1 'JSON(Value|Data)|Json(Value|Data)|Union\[.*None|dict\[str, .*\]\]' "$PWD" --glob '!backend/app/**' --glob '!**/__pycache__/**' | head -160
echo
echo "== check if any JsonValue/JSONValue file exists =="
git ls-files | rg '(^|/).*json(_value|_data|value|data).*\.py$|JsonValue|JSONValue' || true
echo
echo "== deterministic json.dumps behavior for dict[str, Any] values =="
python3 - <<'PY'
import json
from typing import Any
values = [None, True, 0, "", [], {}, object(), lambda x: x]
for v in values:
try:
json.dumps({"credentials": {"key": v}})
except Exception as exc:
print(type(v).__name__, "rejected:", type(exc).__name__, str(exc))
else:
print(type(v).__name__, "serialized")
PYRepository: ProjectTech4DevAI/kaapi-backend
Length of output: 5998
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== JSONValue aliases in tracked files =="
rg -n 'JSON(Value|Data)|Json(Value|Data)|JSONValue|JsonValue|Literal\[(true|null|false)|TypeAlias' --glob '*.py' . | head -200
echo
echo "== deterministic json.dumps behavior for narrow Any values =="
python3 - <<'PY'
import json
from typing import Any, Literal
values = [None, True, 0, "", [], {}, object(), lambda x: x]
for v in values:
try:
json.dumps({"credentials": {"key": v}})
except Exception as exc:
print(type(v).__name__, "rejected:", type(exc).__name__, str(exc))
else:
print(type(v).__name__, "serialized")
for raw in ["true", "null", "false"]:
try:
json.dumps(raw, indent=2)
except Exception as exc:
print(raw, "rejected:", type(exc).__name__, str(exc))
else:
print(repr(raw), "serialized")
PY
echo
echo "== security decryption return type context =="
sed -n '240,290p' backend/app/core/security.pyRepository: ProjectTech4DevAI/kaapi-backend
Length of output: 210
Use a narrow JSON value type for credentials.
dict[str, Any] accepts values such as callables or custom objects that json.dumps rejects at runtime before encryption. Define a recursive JSON value alias and apply it to both credential encryption and decryption signatures so all non-serializable values fail earlier in the crypto contract.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@backend/app/core/security.py` at line 235, Define a recursive JSON value type
alias in security.py and replace dict[str, Any] with the corresponding
JSON-object type in both encrypt_fernet and decrypt_fernet signatures. Ensure
the annotations permit only JSON-serializable primitive, array, and object
values so invalid credentials are rejected at the crypto contract boundary.
Source: Coding guidelines
3 participants
Issue
Closes #1114
Summary
Checklist
Before submitting a pull request, please ensure that you mark these task.
fastapi run --reload app/main.pyordocker compose upin the repository root and test.Notes
Please add here if any other information is required for the reviewer.