Implementation:Apache Kafka GPG Key Validation
| Knowledge Sources | |
|---|---|
| Domains | Release_Engineering, Security |
| Last Updated | 2026-02-09 12:00 GMT |
Overview
Concrete tool for validating GPG key existence and passphrase correctness provided by the Kafka release module.
Description
The gpg module in the Kafka release toolkit provides functions to check whether a GPG key exists in the local keyring and whether a given passphrase is valid for that key. The passphrase validation works by performing a trial signature on a known file, then verifying the resulting signature. If the GPG agent is running, it is killed first to ensure the supplied passphrase is actually tested rather than a cached one.
Usage
Import these functions when building a release automation pipeline that requires GPG signing. Call key_exists first to confirm key availability, then valid_passphrase to test the passphrase before proceeding with artifact signing.
Code Reference
Source Location
- Repository: Apache Kafka
- File: release/gpg.py
- Lines: L27-90
Signature
def key_exists(key_id):
"""
Checks whether the specified GPG key exists locally.
"""
def valid_passphrase(key_id, passphrase):
"""
Checks whether the given passphrase is workable for the given key.
Kills GPG agent first to ensure passphrase is actually tested.
"""
def sign(key_id, passphrase, content, target):
"""
Generates a GPG signature using the given key and passphrase.
"""
def verify(content, signature):
"""
Verify the given GPG signature for the specified content.
"""
Import
from gpg import key_exists, valid_passphrase
I/O Contract
Inputs
| Name | Type | Required | Description |
|---|---|---|---|
| key_id | str | Yes | GPG key identifier to check |
| passphrase | str | Yes (for valid_passphrase) | GPG passphrase to validate |
Outputs
| Name | Type | Description |
|---|---|---|
| key_exists returns | bool | True if the key is in the local GPG keyring |
| valid_passphrase returns | bool | True if the passphrase works for the specified key |
Usage Examples
Verify GPG Prerequisites
from gpg import key_exists, valid_passphrase
key_id = "ABCDEF1234567890"
passphrase = "my-secret-passphrase"
# Step 1: Check key exists
if not key_exists(key_id):
print(f"GPG key {key_id} not found in local keyring")
sys.exit(1)
# Step 2: Validate passphrase
if not valid_passphrase(key_id, passphrase):
print("Invalid GPG passphrase")
sys.exit(1)
print("GPG prerequisites verified successfully")