# frozen_string_literal: true # adn/tools/cli/nodos/info.rb — Utilidades para extracción de metadatos de fichas de nodos # ====================================================================================== # # Fase 2.5: Soporte dual — Tabla canónica (META:BEGIN/META:END) + fallback regex # # Prioridad de lectura: # 1. Tabla canónica entre y # 2. Regex heurísticas (legacy, para fichas no migradas) require_relative '../../core/constants' module ADN module NodosInfo def self.extraer_metadata(nodo_nombre) ruta = File.join(ADN::NODOS_DIR, "#{nodo_nombre}.md") return nil unless File.exist?(ruta) contenido = File.read(ruta, encoding: 'UTF-8') # ─── Intentar tabla canónica primero ────────────────────── meta = extraer_meta_canonica(contenido) if meta return { nombre: nodo_nombre, ip: meta['IP'], ip_lan: detectar_ip_lan(contenido), host: meta['Host'], so: detectar_so_desde_valor(meta['OS']), vmid: meta['VMID'], puerto_ssh: extraer_puerto_de_ssh(meta['SSH']), usuario: extraer_usuario_de_ssh(meta['SSH']), clave_boveda: meta['Bóveda'], auth_type: meta['Auth']&.to_sym, clave_ssh: meta['Clave SSH'], ssh_opts: meta['SSH Opts'], ring: meta['Ring'], estado: meta['Estado'], _fuente: :canonica } end # ─── Fallback: Regex heurísticas (legacy) ───────────────── { nombre: nodo_nombre, ip: detectar_ip(contenido), ip_lan: detectar_ip_lan(contenido), host: detectar_host(contenido), so: detectar_so(contenido), vmid: detectar_vmid(contenido), puerto_ssh: detectar_puerto_ssh(contenido) || 22, usuario: detectar_usuario(contenido), clave_boveda: detectar_clave_boveda(contenido), auth_type: detectar_auth_type(contenido), clave_ssh: detectar_clave_ssh(contenido), ssh_opts: detectar_ssh_opts(contenido), ring: nil, estado: nil, _fuente: :regex } end def self.extraer_ip_lan(nodo_nombre) ruta = File.join(ADN::NODOS_DIR, "#{nodo_nombre}.md") return nil unless File.exist?(ruta) contenido = File.read(ruta, encoding: 'UTF-8') detectar_ip_lan(contenido) end # ─── Parser de tabla canónica ────────────────────────────── def self.extraer_meta_canonica(contenido) match = contenido.match(/(.*?)/m) return nil unless match meta = {} match[1].each_line do |linea| # | **Campo** | `valor` | o | **Campo** | valor | if linea =~ /\|\s*\*\*(.+?)\*\*\s*\|\s*`?([^`|]+?)`?\s*\|/ campo = $1.strip valor = $2.strip meta[campo] = valor unless valor.empty? end end meta.empty? ? nil : meta end def self.extraer_usuario_de_ssh(ssh_str) return nil unless ssh_str # Formato: user@ip:port if ssh_str =~ /^([a-zA-Z0-9_\-\.]+)@/ return $1 end nil end def self.extraer_puerto_de_ssh(ssh_str) return 22 unless ssh_str # Formato: user@ip:port if ssh_str =~ /:(\d+)$/ return $1.to_i end 22 end def self.detectar_so_desde_valor(os_str) return :linux unless os_str return :linux if os_str =~ /linux|ubuntu|debian|centos|rocky|proxmox|xen/i return :windows if os_str =~ /windows|server|win/i return :firmware if os_str =~ /firmware|hikvision|dvr|mikrotik|routeros/i :linux end # ─── Regex heurísticas (legacy — para fichas sin META) ───── def self.detectar_usuario(contenido) # **SSH**: user@ip:port o DOMAIN\user@ip:port if contenido =~ /\*\*SSH\*\*:\s*`?([^`\s(]+)/i full = $1 # Extraer usuario de formato user@ip:port o DOMAIN\user@ip:port if full.include?('@') return full.split('@')[0...-1].join('@') if full.split('@').last =~ /^[\d\.]+/ return full.split(':')[0] end return full end # Fallback: | SSH | user@domain@ip:7022 | if contenido =~ /(?:SSH|Acceso)\*\*?:?\s*`?([a-zA-Z0-9_\-\.\\@]+)/i full = $1 if full.include?('@') parts = full.split('@') if parts.last =~ /^\d{1,3}(\.\d{1,3}){3}/ return parts[0...-1].join('@') else return full.split(':')[0] end end return full end nil end def self.detectar_ip(contenido) # Prioridad 1: IP de línea SSH (es la IP operativa para conexión) if contenido =~ /\*\*SSH\*\*:\s*`?[^@]+@([\d\.]+)/i return $1 end # | IP / Ubicación | 10.0.10.10 | # | IP | 10.0.100.11/24 | if contenido =~ /\|\s*IP\s*(?:\/\s*Ubicación)?\s*\|\s*`?([\d\.]+)/i return $1 end # - **IP**: `10.0.10.200` if contenido =~ /IP\*\*:\s*`?([\d\.]+)/i return $1 end nil end def self.detectar_ip_lan(contenido) # Buscar IP de LAN en formato: `10.0.100.1` (después de "LAN" o "Bridge" o "Alias") if contenido =~ /(?:LAN|Bridge vmbr0|Alias).*?`([\d\.]+)`/i return $1 end # - **IP**: `10.0.100.12/24` (Estática) - buscar solo la IP antes de / if contenido =~ /\*\*IP\*\*:\s*`([\d\.]+)\// return $1 end nil end def self.detectar_host(contenido) if contenido =~ /\*\*(?:Padre\/Host|Host Anfitrión|Anfitrión|Host)\*\*\s*[:\|]?\s*`?([a-zA-Z0-9\-]+)`?/i return $1 end nil end def self.detectar_so(contenido) # | Sistema Operativo | ... | o - **Sistema Operativo**: ... if contenido =~ /(?:\||\-)\s*\*\*?(?:Sistema Operativo|SO|OS)\*\*?\s*[:\|]\s*([^\|\n]+)/i so_raw = $1.strip return :linux if so_raw =~ /linux|ubuntu|debian|centos|rocky|proxmox|xen/i return :windows if so_raw =~ /windows|server|win/i end :linux # Default end def self.detectar_vmid(contenido) if contenido =~ /(?:VM ID|vmid|VMID)\*\*?:?\s*`?(\d+)`?/i return $1 end nil end def self.detectar_puerto_ssh(contenido) # **SSH**: user@ip:7022 if contenido =~ /\*\*SSH\*\*:\s*`?[^`\s]+:(\d+)/i return $1.to_i end # | SSH | user@ip:7022 | if contenido =~ /(?:SSH|Puerto|Puerto SSH)\*\*?:?\s*.*:(\d+)/i return $1.to_i end # (Puerto 7022) if contenido =~ /\(Puerto\s+(\d+)\)/i return $1.to_i end nil end # Detecta la clave de la bóveda de candados desde la ficha # Formato esperado: (Bóveda: `clave`) o (Bóveda: clave) def self.detectar_clave_boveda(contenido) if contenido =~ /\*\*SSH\*\*:.*Bóveda:\s*`?([a-zA-Z0-9_\-:]+)`?\)?/i return $1 end # Fallback: buscar en cualquier línea de credenciales if contenido =~ /[Bb]óveda:\s*`?([a-zA-Z0-9_\-:]+)`?\)?/ return $1 end nil end def self.detectar_auth_type(contenido) # Detectar 'RSA Legacy' para XenServer 7.0 (OpenSSH 6.6.1) if contenido =~ /RSA Legacy/i return :rsa_legacy end # Detectar otros tipos de autenticación if contenido =~ /\*\*SSH\*\*:.*\((Passphrase|Password|RSA)/i match = $1.downcase return :passphrase if match == 'passphrase' return :rsa if match == 'rsa' return :password end :password # Default para VMs Windows end # Detectar clave SSH específica para el nodo def self.detectar_clave_ssh(contenido) # Buscar en formato: **Clave SSH**: `~/.ssh/id_rsa_xen` if contenido =~ /\*\*Clave SSH\*\*:\s*`?([~a-zA-Z0-9_\-\.\/]+)`?/i return $1 end # Fallback: buscar en línea de comando SSH if contenido =~ /ssh -i ([~a-zA-Z0-9_\-\.\/]+)\s/i return $1 end nil end # Detectar opciones SSH especiales (para servidores legacy) def self.detectar_ssh_opts(contenido) # Buscar opciones -o PubkeyAcceptedAlgorithms o -o HostkeyAlgorithms if contenido =~ /(-o PubkeyAcceptedAlgorithms=[^\s`]+)\s+(-o HostkeyAlgorithms=[^\s`]+)/i return "#{$1} #{$2}" end nil end end end