Airflow使用SSHHook传入StringIO格式PEM密钥触发文件不存在报错

报错和私钥换行、StringIO的newline参数没有任何关系，核心是SSHHook的key_file参数仅接受本地私钥文件的路径字符串，不支持传入StringIO类文件对象。
从报错栈可以看到Paramiko拿到传入值后，直接调用open(filename, "r")读取私钥，你传入的StringIO对象被转成字符串后，内容刚好是私钥开头的-----BEGIN RSA PRIVATE KEY-----\n，Paramiko把这段字符串当成了文件路径去查找，自然会抛出文件不存在的错误。

你可以根据自己安装的Airflow SSH Provider版本选对应方案：

• 方案1：临时落盘私钥（全版本兼容）
  把从Secrets Manager获取的私钥写入权限为600的临时文件，将临时文件路径传入key_file参数，执行完逻辑后自动删除临时文件，避免私钥残留：

  import tempfile
  import os
  from airflow.providers.ssh.hooks.ssh import SSHHook
  from airflow.providers.amazon.aws.hooks.secrets_manager import SecretsManagerHook
  
  secrets_manager_hook = SecretsManagerHook()
  sm_client = secrets_manager_hook.get_conn()
  secret = sm_client.get_secret_value(SecretId='<SECRET>')
  pem_key_value = secret["SecretString"]
  
  # 创建临时文件存储私钥
  with tempfile.NamedTemporaryFile(mode='w', suffix='.pem', delete=False) as pem_file:
      pem_file.write(pem_key_value)
      temp_pem_path = pem_file.name
  # 私钥权限必须设为600，否则会被SSH判定为不安全拒绝加载
  os.chmod(temp_pem_path, 0o600)
  
  try:
      ssh_hook = SSHHook(
          ssh_conn_id=None,
          remote_host='<HOST>',
          username='ec2-user',
          key_file=temp_pem_path
      )
      ssh_conn = ssh_hook.get_conn()
      stdin, stdout, stderr = ssh_hook.exec_ssh_client_command(
          ssh_client=ssh_conn,
          command='echo Hello',
          get_pty=False
      )
      print(stdout.read().decode())
  finally:
      # 执行完成后删除临时私钥
      os.unlink(temp_pem_path)
  

• 方案2：直接传入私钥内容（无落盘，适用于apache-airflow-providers-ssh >= 3.0.0）
  高版本SSHHook新增了private_key参数，直接接收私钥原文字符串，不需要转StringIO、不需要写临时文件，代码更简洁：

  from airflow.providers.ssh.hooks.ssh import SSHHook
  from airflow.providers.amazon.aws.hooks.secrets_manager import SecretsManagerHook
  
  secrets_manager_hook = SecretsManagerHook()
  sm_client = secrets_manager_hook.get_conn()
  secret = sm_client.get_secret_value(SecretId='<SECRET>')
  pem_key_value = secret["SecretString"]
  
  ssh_hook = SSHHook(
      ssh_conn_id=None,
      remote_host='<HOST>',
      username='ec2-user',
      private_key=pem_key_value
  )
  ssh_conn = ssh_hook.get_conn()
  stdin, stdout, stderr = ssh_hook.exec_ssh_client_command(
      ssh_client=ssh_conn,
      command='echo Hello',
      get_pty=False
  )
  print(stdout.read().decode())
  

• 不需要手动给私钥每行追加\n，只要从Secrets Manager拿到的私钥内容和本地原始PEM文件换行格式完全一致，就可以直接使用
• 禁止给key_file参数传入文件对象、StringIO对象，该参数的类型定义就是字符串格式的本地文件路径
• 如果使用临时文件方案，一定要把文件权限设为0o600，否则Paramiko/SSH会因为私钥权限过宽拒绝加载

内容的提问来源于stack exchange，提问作者rk92