Skip to content

feat(azure): Support x509 certificate SSH keys from IMDS#6912

Open
peytonr18 wants to merge 13 commits into
canonical:mainfrom
peytonr18:probertson-imds-x509
Open

feat(azure): Support x509 certificate SSH keys from IMDS#6912
peytonr18 wants to merge 13 commits into
canonical:mainfrom
peytonr18:probertson-imds-x509

Conversation

@peytonr18

@peytonr18 peytonr18 commented Jun 25, 2026

Copy link
Copy Markdown
Contributor

Proposed Commit Message

feat(azure): Support x509 certificate SSH keys from IMDS

Azure IMDS may return SSH public keys in OpenSSH format or as x509
certificates. Non-OpenSSH keys previously forced a fallback to
Wireserver. _get_public_keys_from_imds() now converts x509 certificates
to OpenSSH format (via the certs module) and uses them directly.
Unsupported formats and conversion failures still fall back gracefully.

Changes

  • DataSourceAzure.py: _get_public_keys_from_imds() now handles keys
    per-item — OpenSSH keys used as-is, x509 certs converted, other
    formats raise ValueError (preserving Wireserver fallback).
  • tests/.../test_azure.py: added tests for single x509, mixed
    x509 + OpenSSH, and conversion-failure fallback.

Notes

Merge type

  • Squash merge using "Proposed Commit Message"

peytonr18 added 11 commits June 23, 2026 14:36
…sity

Added extract_x509_certificate() to validate certificates in bundles, integrated validation into parse_certificates(), and toned down debug logs to avoid sounding like failures.
Fold the regex-based certificate parsing into cloudinit.sources.azure.certs so
helpers no longer reimplement it, and wire OpenSSLManager.parse_certificates to
loop over the helper. Added a regression test that confirms CRLF-mixed bundles
still yield every fingerprint + key pair.
…all_certificates to avoid openssl dependency
…and iterate

them directly in parse_certificates, replacing the previous loop/find slicing flow.

Also clarify is_openssh_formatted behavior by explicitly rejecting embedded CRLF
and improving debug messages; update tests and
type hints to match the new extraction API.
Azure-generated SSH keys may contain \r\n sequences embedded in the
base64 key data (LP: #1910835). Previously, keys with embedded CRLF
were rejected outright by is_openssh_formatted(). Instead, sanitize
the keys by stripping \r\n before validation so they can be properly
written to authorized_keys.
- Use pytest.mark.parametrize, fixtures, and monkeypatch in tests
- Remove raw docstring prefix; clarify CRLF deprecation in docstring
- Remove redundant _key_is_openssh_formatted wrapper
- Simplify sanitize_openssh_key call site and log messages
- Reuse variables and constants instead of inline string literals
Azure IMDS may deliver SSH public keys in OpenSSH format or as x509
certificates. Previously any non-OpenSSH key caused
_get_public_keys_from_imds() to raise and fall back to Wireserver.

Inspect each IMDS key individually: keep OpenSSH-formatted keys,
convert x509 certificates to OpenSSH format via the certs module, and
raise ValueError for unsupported formats (preserving the Wireserver
fallback). Conversion failures (openssl/ssh-keygen) are caught and
re-raised as ValueError so a malformed certificate falls back
gracefully instead of crashing metadata crawl.

Adds unit tests for single x509, mixed x509 + OpenSSH, and
conversion-failure cases.
@peytonr18
peytonr18 marked this pull request as ready for review July 23, 2026 18:54
# Conflicts:
#	cloudinit/sources/DataSourceAzure.py
Comment thread cloudinit/sources/DataSourceAzure.py Outdated
Comment on lines +1066 to +1094
@@ -1077,12 +1080,24 @@ def _get_public_keys_from_imds(self, imds_md: dict) -> List[str]:
report_diagnostic_event(log_msg, logger_func=LOG.debug)
raise

ssh_keys = [certs.sanitize_openssh_key(key) for key in ssh_keys]

if any(not certs.is_openssh_formatted(key) for key in ssh_keys):
log_msg = "Key(s) not in OpenSSH format"
report_diagnostic_event(log_msg, logger_func=LOG.debug)
raise ValueError(log_msg)
ssh_keys = []
for key in raw_keys:
sanitized = certs.sanitize_openssh_key(key)
if certs.is_openssh_formatted(sanitized):
ssh_keys.append(sanitized)
elif certs.is_x509_certificate(key):
log_msg = "Converting x509 certificate from IMDS to OpenSSH"
report_diagnostic_event(log_msg, logger_func=LOG.debug)
try:
ssh_keys.append(certs.convert_x509_to_openssh(key).strip())
except subp.ProcessExecutionError as error:
log_msg = "Failed to convert x509 certificate from IMDS"

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This question is purely theoretical, but would there be additional value in logging the item that prompted the failure in conversion? If many certificates get specified (I forget the max number allowed off the top of my head), and one fails in the middle, it could make that case easier to determine which failed.

I do not know much about these certificates, but I believe they are public so there should not be any risk in logging them (this would need to be validated). One other consideration is that the failed certificate may be too long for the standard diagnostic event report, although I haven't tested how long these events can be. If either of those two thoughts get in the way of logging the value, we can easily resolve this comment and err on the safe side.

Other than that little thought, this looks awesome! Great work.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I like this idea! You're right that these are public certs so logging isn't a leak risk, so I added this suggestion to the latest commit.

Comment thread cloudinit/sources/DataSourceAzure.py Outdated
"""
try:
ssh_keys = [
raw_keys = [

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

perhaps public_keys?

Comment thread cloudinit/sources/DataSourceAzure.py Outdated
ssh_keys = []
for key in raw_keys:
sanitized = certs.sanitize_openssh_key(key)
if certs.is_openssh_formatted(sanitized):

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

if you roll this up into something like normalize_ssh_public_key()

ssh_keys = [certs.normalize_ssh_public_key(key) for key in ssh_keys]

Move the IMDS public key sanitize/convert logic into
certs.normalize_ssh_public_key() and log the offending key on failure.
Add unit tests.
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

3 participants