Cloudera Data Engineering(CDE) 에서 DB 비밀번호 같은 값을 다룰 때 두 조건을 동시에 만족해야 했다.
추가 제약으로 런타임에 패키지를 설치할 수 없고, Spark job 은 PySpark 이며 Scala 는 선택지에서 제외됐다.
CDE 1.5.0 에 도입돼 최신(on-prem 1.24.x, Cloud) 까지 표준으로 유지되는 기능이다. 잡 정의나 코드에 자격증명을 하드코딩하는 대신 CDE 에 암호화 저장하고 런타임에 read-only 파일로 마운트한다.
# 값은 프롬프트에서 숨김 입력되므로 셸 히스토리에 남지 않는다
cde credential create \
--name workload-cred-1 \
--type workload-credential \
--workload-cred-key db-pass \
--workload-cred-key aws-secret
cde session create --name s1 --type pyspark --workload-credential workload-cred-1
마운트 경로는 Spark driver · executor 의 /etc/dex/secrets/<credential-name>/<key> 다.
with open("/etc/dex/secrets/workload-cred-1/db-pass") as f:
db_pass = f.read().strip()
jdbcDF = spark.read.jdbc(
"jdbc:postgresql://<db-host>:5432/<db>",
"schema.tablename",
properties={"user": "<db-user>", "password": db_pass},
)
--workload-cred-json-file 로 JSON 파일을 쓸 수도 있지만 파일에 평문이 남으므로 사용 후 즉시 파기한다.
Workload Secret 은 Spark driver/executor 에만 마운트된다. CDE 는 Virtual Cluster 마다 Airflow 를 내장하고, Private Cloud 는 CDE job run operator 만 지원하며 Airflow provider package 는 지원하지 않는다(트러블슈팅 시 제거를 요구할 수 있다).
Airflow Pod (DAG) — workload secret 파일 마운트 안 됨
└ CDEJobRunOperator 로 job 트리거만
Spark Driver/Executor Pod — /etc/dex/secrets/<cred>/<key> 마운트
그래서 방식별로 두 조건의 충족 여부가 갈린다.
| 방식 | 조건1 UI 평문 없음 | 조건2 동일 값 공유 |
|---|---|---|
| Workload Secret 만 | 충족 | DAG 는 값을 쓰지 않으므로 공유가 아님 |
Airflow Variable → variables/overrides 로 전달 |
위반 위험 | 충족 |
| Airflow Connection + Workload Secret 분리 | 충족 | 두 저장소를 수동 동기화해야 해 단일 소스가 아님 |
| 외부 Secrets Backend(Vault 등) | 충족 | 충족하지만 미지원 provider package 의존 |
두 번째 방식이 표준처럼 보이지만, variables={'key': '{{ var.value.key }}'} 나 overrides 로 넘긴 값은 Spark job 실행 설정이 되어 CDE 잡 실행 화면 · Spark UI · Airflow rendered template 에 평문으로 비칠 수 있다. Airflow 의 마스킹은 키 이름 기반 + 값 매칭 기반이라 임의 conf 주입까지 보장하지 않는다.
대칭키를 쓰면 같은 키를 Airflow 쪽과 Spark 쪽 양쪽에 둬야 해 비밀이 두 곳으로 복제된다. 키쌍으로 나누면 비밀(개인키) 은 Spark 한 곳에만 남는다.
[1회 준비]
개인키 → cde credential (workload-credential), Spark 에만 마운트
공개키 → DAG 코드 / Airflow Variable / Resource. 비밀이 아니라 노출돼도 무방
[매 실행]
DAG: 원본 V 를 공개키로 암호화 → 암호문 C
CDEJobRunOperator(overrides={'spark':{'conf':{'app.enc.payload': C}}})
(UI 에는 C 만 보인다)
Spark: conf 에서 C 를 읽고 /etc/dex/secrets/<cred>/<key> 의 개인키로 복호화 → V
CDE 1.25 는 Airflow 2.9 / Python 3.11 / Spark 3.5 / Java 17 이다.
cryptography 로 구동되므로 이 패키지가 이미 존재한다. DAG 에서 RSA-OAEP 암호화가 가능하다.cryptography 가 포함된다는 보장이 없다(실제로 없음이 확인됐다). 대신 PySpark 는 py4j 로 JVM 에 접근할 수 있으므로 JDK 내장 JCE(javax.crypto) 를 호출하면 추가 설치 0개로 복호화된다.순수 Python 대안으로 검토한 pyaes 는 AES 대칭키 전용이라 비대칭 요구에 원천적으로 맞지 않는다. 마지막 릴리스가 2017년(1.6.1) 이고 GCM(AEAD) 도 없어 무결성 검증이 빠진다. CDE files resource 에 소스를 올려 sys.path 에 추가하면 import 자체는 되지만, 얻는 것이 대칭 AES 뿐이라 키가 두 곳에 복제되는 원래 문제로 돌아간다.
CDE Sessions(pyspark 타입) 에서 driver 의 spark._jvm 으로 검증했다.
# Stage 1 — JCE 접근 확인
jvm = spark._jvm
print("Java version:", jvm.java.lang.System.getProperty("java.version"))
print("Cipher OK :", jvm.javax.crypto.Cipher
.getInstance("RSA/ECB/OAEPWithSHA-256AndMGF1Padding").getAlgorithm())
# → Java version: 17.0.15-internal
# → Cipher OK : RSA/ECB/OAEPWithSHA-256AndMGF1Padding
# [오프라인 = DAG 역할] 키쌍 생성 + 공개키 암호화
from cryptography.hazmat.primitives.asymmetric import rsa, padding
from cryptography.hazmat.primitives import hashes, serialization
import base64
key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
priv_der = key.private_bytes(
serialization.Encoding.DER,
serialization.PrivateFormat.PKCS8,
serialization.NoEncryption(),
)
sample = "test-secret-value-123"
ct = key.public_key().encrypt(
sample.encode("utf-8"),
# Java 의 OAEPWithSHA-256AndMGF1Padding 은 MGF1 이 SHA-1 이 기본값이므로 맞춰 준다
padding.OAEP(mgf=padding.MGF1(hashes.SHA1()), algorithm=hashes.SHA256(), label=None),
)
print("PRIV_B64_LEN =", len(base64.b64encode(priv_der).decode()))
print("CT_B64_LEN =", len(base64.b64encode(ct).decode()))
# [CDE Session = Spark 역할] 개인키 로드 + 복호화
jvm = spark._jvm
with open("/etc/dex/secrets/rsa-dec-key/rsa-priv") as f:
PRIV_B64 = f.read().strip()
CT_B64 = spark.conf.get("app.enc.payload")
der = jvm.java.util.Base64.getDecoder().decode(PRIV_B64)
spec = jvm.java.security.spec.PKCS8EncodedKeySpec(der)
priv = jvm.java.security.KeyFactory.getInstance("RSA").generatePrivate(spec)
dec = jvm.javax.crypto.Cipher.getInstance("RSA/ECB/OAEPWithSHA-256AndMGF1Padding")
dec.init(jvm.javax.crypto.Cipher.DECRYPT_MODE, priv)
V = jvm.java.lang.String(dec.doFinal(jvm.java.util.Base64.getDecoder().decode(CT_B64)), "UTF-8")
# V 는 절대 print · 로그에 남기지 않는다
개인키는 길어서 대화형 입력이 어려우므로 JSON 파일로 등록하고 즉시 파기한다.
cat > /tmp/wc.json << 'EOF'
{"rsa-priv": "<PRIV_B64 한 줄>"}
EOF
cde credential create --name rsa-dec-key --type workload-credential --workload-cred-json-file /tmp/wc.json
shred -u /tmp/wc.json
cde session create --name dec-test --type pyspark --workload-credential rsa-dec-key
검증 결과 recovered: test-secret-value-123, 즉 DAG 측 cryptography 암호문을 PySpark driver 의 JVM JCE 로 추가 패키지 없이 복호화하는 경로가 성립한다.
KeyPairGenerator.getInstance("RSA").generateKeyPair() 를 py4j 로 호출하면 JDK 17 의 모듈 강캡슐화 때문에 InaccessibleObjectException: module java.base does not "opens java.security" 가 난다. 반환 객체가 비공개 클래스 KeyPairGenerator$Delegate 이기 때문이다. 실제 설계에서 Spark 는 키를 생성하지 않고 KeyFactory · Cipher 같은 공개 클래스로 로드 · 복호화만 하므로 이 경로에 걸리지 않는다.spark._jvm 게이트웨이는 executor 의 lambda(map/mapPartitions) 안에서는 쓸 수 없다. driver 에서 1회 복호화하는 용도라면 문제가 없지만 행 단위 복호화가 필요하면 이 경로는 막힌다.byte[] 를 직접 넘기지 말고 base64 문자열만 오가게 한다. JVM 안에서 java.util.Base64 로 디코딩한다.... 로 잘린다. Illegal base64 character 2e(0x2e 는 .) 는 복사 사고의 신호다. 오프라인에서 길이를 함께 출력하고 세션 쪽에서 길이 · 허용 문자를 검증하는 가드를 둔다.OAEPParameterSpec 를 명시해야 한다.원본 값의 단일 소스를 Airflow Connection 에 둘지 외부 저장소에 둘지, OAEP 파라미터 통일, 페이로드 크기 대응, DAG 암호화를 런타임(PythonOperator) 에 할지 오프라인 1회로 할지는 결정 전에 대화가 끝났다.