PGP Encrypt & Decrypt — Ruby source

# PGP Encrypt & Decrypt — read OpenPGP key metadata, encrypt (optionally
# signing), decrypt.
#
# Language: Ruby (3.x, standard library only — open3/tmpdir/fileutils drive the
#           `gpg` binary, the same engine the C port reaches through GPGME and
#           the native counterpart to the TS reference's openpgp.js)
# Source:   CosmoDev polyglot showcase port of the PGP Encrypt & Decrypt tool,
#           ported from src/lib/pgp-encrypt.ts (the canonical TypeScript
#           implementation).
# License:  display source — part of CosmoDev's polyglot tool pages.
#
# Three entry points, matching the TS public API one-for-one:
#   read_key_info — user ID, uppercase fingerprint, algorithm, creation date,
#                   expiry (nil when the key never expires), and whether the
#                   armor holds a private key.
#   pgp_encrypt   — ASCII-armored PGP message for a recipient's public key,
#                   optionally signed with the sender's private key.
#   pgp_decrypt   — plaintext from an armored message + private key.
#
# Armored key material is imported into a fresh 0700 GNUPGHOME under a temp dir
# that is deleted on the way out, so nothing is written to the user's real
# keyring — the Ruby equivalent of the TS island keeping every operation
# client-side.
#
# gpg is always spawned with an ARGUMENT ARRAY, never a command string, so no
# key, message or passphrase is ever parsed by a shell. Passphrases go in on
# stdin via --passphrase-fd 0 rather than argv, where any local user could read
# them out of the process table.

require 'fileutils'
require 'open3'
require 'tmpdir'

module PGPEncrypt
  # Key metadata, mirroring PGPKeyInfo in the TS reference.
  KeyInfo = Struct.new(:user_id, :fingerprint, :algorithm, :creation_date,
                       :expiry, :private, keyword_init: true) do
    # `private` collides with Ruby's visibility keyword when read bare.
    alias_method :private?, :private
  end

  PRIVATE_ARMOR_HEADERS = [
    '-----BEGIN PGP PRIVATE KEY BLOCK-----',
    '-----BEGIN PGP SECRET KEY BLOCK-----'
  ].freeze

  # gpg's --with-colons public-key algorithm numbers (RFC 4880 §9.1 + the
  # modern additions) mapped to the names openpgp.js reports.
  ALGORITHM_NAMES = {
    '1' => 'rsaEncryptSign', '2' => 'rsaEncrypt', '3' => 'rsaSign',
    '16' => 'elgamal', '17' => 'dsa', '18' => 'ecdh', '19' => 'ecdsa',
    '22' => 'eddsaLegacy', '25' => 'x25519', '27' => 'ed25519'
  }.freeze

  module_function

  # Read a PGP key (public or private) and extract its metadata.
  # Raises on invalid or unrecognized key material.
  def read_key_info(armored_key)
    trimmed = armored_key.to_s.strip
    raise ArgumentError, 'Key input is empty or invalid.' if trimmed.empty?

    # Detect key type from the armor header, exactly as the TS reference does —
    # before handing anything to gpg.
    is_private = PRIVATE_ARMOR_HEADERS.any? { |header| trimmed.include?(header) }

    with_ephemeral_home do |home|
      import_key(home, trimmed)
      pub = colon_record(home, 'pub') or raise 'Invalid PGP key: no primary key packet.'

      KeyInfo.new(
        user_id: first_user_id(home),
        fingerprint: primary_fingerprint(home).upcase,
        algorithm: ALGORITHM_NAMES.fetch(pub[3], "unknown(#{pub[3]})"),
        creation_date: epoch_to_date(pub[5]),
        expiry: epoch_to_date(pub[6]), # blank field = never expires -> nil
        private: is_private
      )
    end
  rescue ArgumentError
    raise
  rescue StandardError => e
    raise "Invalid PGP key: #{e.message}"
  end

  # Encrypt a plaintext message for a recipient's public key. Returns an
  # ASCII-armored PGP message. When `signing_private_key` is given the message
  # is signed as well; `passphrase` unlocks that signing key.
  def pgp_encrypt(message, public_key, signing_private_key: nil, passphrase: nil)
    raise ArgumentError, 'Message must not be empty.' if message.to_s.empty?
    raise ArgumentError, 'Recipient public key must not be empty.' if public_key.to_s.empty?

    pass = passphrase.to_s.empty? ? nil : passphrase.to_s

    with_ephemeral_home do |home|
      import_key(home, public_key.to_s.strip)
      recipient = primary_fingerprint(home)

      args = ['--armor', '--encrypt', '--recipient', recipient,
              # The ephemeral keyring carries no ownertrust, so every recipient
              # would otherwise trip gpg's "not certified" prompt.
              '--trust-model', 'always']

      if signing_private_key.to_s.strip.empty?
        run_gpg(home, args, stdin: message.to_s)
      else
        import_key(home, signing_private_key.to_s.strip, passphrase: pass)
        signer = secret_fingerprint(home)
        # --passphrase-fd 0 is taken by the passphrase, so the plaintext is fed
        # through a temp file inside the ephemeral (0700) home instead.
        encrypt_signed(home, args, message.to_s, signer, pass)
      end
    end
  end

  # Decrypt an ASCII-armored PGP message with the recipient's private key.
  # Returns the plaintext.
  def pgp_decrypt(armored_message, private_key, passphrase: nil)
    raise ArgumentError, 'Armored message must not be empty.' if armored_message.to_s.empty?
    raise ArgumentError, 'Private key must not be empty.' if private_key.to_s.empty?

    pass = passphrase.to_s.empty? ? nil : passphrase.to_s

    with_ephemeral_home do |home|
      import_key(home, private_key.to_s.strip, passphrase: pass)
      message_path = write_temp(home, 'message.asc', armored_message.to_s.strip)
      run_gpg(home, ['--decrypt', message_path], passphrase: pass)
    end
  end

  # --- internals -------------------------------------------------------------

  # A private keyring that exists only for the duration of the block. 0700 is
  # required — gpg refuses to run against a world-readable home.
  def with_ephemeral_home
    dir = Dir.mktmpdir('cosmodev-pgp-')
    FileUtils.chmod(0o700, dir)
    yield dir
  ensure
    FileUtils.remove_entry(dir) if dir && File.directory?(dir)
  end

  # Spawn gpg with an argument array (never a shell string) against `home`.
  # Exactly one of `stdin` / `passphrase` may occupy stdin; a passphrase wins,
  # so callers that need both route the payload through a temp file.
  def run_gpg(home, args, stdin: nil, passphrase: nil)
    base = ['gpg', '--homedir', home, '--batch', '--no-tty', '--yes']
    base += ['--pinentry-mode', 'loopback', '--passphrase-fd', '0'] if passphrase
    input = passphrase ? "#{passphrase}\n" : stdin.to_s

    out, err, status = Open3.capture3(*base, *args, stdin_data: input)
    raise(err.strip.empty? ? "gpg exited #{status.exitstatus}" : err.strip) unless status.success?

    out
  end

  # Encrypt + sign. The plaintext goes to a file so stdin stays free for the
  # signing key's passphrase.
  def encrypt_signed(home, args, message, signer, pass)
    path = write_temp(home, 'plaintext.txt', message)
    run_gpg(home, args + ['--sign', '--local-user', signer, path], passphrase: pass)
  end

  # Import armored key material. Importing an encrypted secret key needs the
  # passphrase, so it is forwarded when present.
  def import_key(home, armored, passphrase: nil)
    path = write_temp(home, 'key.asc', armored)
    run_gpg(home, ['--import', path], passphrase: passphrase)
  end

  # Write inside the 0700 ephemeral home — never /tmp at large — so key
  # material and plaintext are never world-readable, even briefly.
  def write_temp(home, name, content)
    path = File.join(home, name)
    File.open(path, File::WRONLY | File::CREAT | File::TRUNC, 0o600) { |io| io.write(content) }
    path
  end

  # The first colon record of the given type, split into its fields.
  def colon_record(home, type, secret: false)
    listing = run_gpg(home, [secret ? '--list-secret-keys' : '--list-keys', '--with-colons'])
    row = listing.each_line.find { |line| line.start_with?("#{type}:") }
    row&.split(':')
  end

  def primary_fingerprint(home)
    fpr = colon_record(home, 'fpr') or raise 'no fingerprint record'
    fpr[9].to_s
  end

  def secret_fingerprint(home)
    listing = run_gpg(home, ['--list-secret-keys', '--with-colons'])
    row = listing.each_line.find { |line| line.start_with?('fpr:') }
    raise 'Signing key holds no secret material.' unless row

    row.split(':')[9].to_s
  end

  # uid records carry the user ID in field 10; 'unknown' matches the TS
  # fallback when a key has no user ID at all.
  def first_user_id(home)
    uid = colon_record(home, 'uid')
    value = uid && uid[9].to_s.strip
    value.nil? || value.empty? ? 'unknown' : unescape_colon_field(value)
  end

  # gpg escapes non-ASCII and delimiter bytes in colon output as \xNN.
  def unescape_colon_field(value)
    value.gsub(/\\x([0-9a-fA-F]{2})/) { Regexp.last_match(1).hex.chr }
         .force_encoding(Encoding::UTF_8)
  end

  # Colon-format timestamps are UNIX epoch seconds; an empty field means "no
  # expiry", which the TS reference surfaces as null.
  def epoch_to_date(field)
    return nil if field.nil? || field.to_s.strip.empty?

    Time.at(Integer(field, 10)).utc.strftime('%Y-%m-%d')
  rescue ArgumentError, TypeError
    nil
  end
end