dpl/scripts/node-onboarding/ssh_pool.py
hellcat 5c9157f5d7
Some checks are pending
Docker Build and Publish / build (push) Waiting to run
feat: add reusable jump-host node onboarding runtime
2026-10-17 15:22:30 +08:00

100 lines
5.1 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""Thread-local SSH sessions, mandatory explicit jump, strict known-host keys."""
import json
import hashlib
import base64
import re
import threading
from pathlib import Path
def private_text(path):
p = Path(path)
if not p.is_absolute() or p.is_symlink() or not p.is_file() or p.stat().st_mode & 0o077:
raise ValueError('Private regular absolute credential/config reference required')
return p.read_text()
def private_json(path):
return json.loads(private_text(path))
def credentials(spec):
if not spec.get('credential_ref'):return {}
if spec.get('credential_format','json')=='json':return private_json(spec['credential_ref'])
if spec['credential_format']=='codocs-password':
matches=re.findall(r'(?m)^[ \t]*(?:[-*][ \t]+)?密码[::][ \t]*([^\r\n]+)$',private_text(spec['credential_ref']))
if len(matches)!=1:raise ValueError('Unambiguous existing CoDocs password field required')
return {'password':matches[0].strip()}
raise ValueError('Unsupported credential reference format')
class Pool:
def __init__(self, config, jump_id, client_factory=None):
self.config, self.jump_id = config, jump_id
if jump_id not in config['hosts'] or config['hosts'][jump_id].get('role') != 'white_jump':
raise ValueError('Explicit registered white jump required')
if client_factory is None:
known=Path(config['known_hosts'])
if not known.is_absolute() or known.is_symlink() or not known.is_file() or known.stat().st_mode & 0o022:raise ValueError('Explicit trusted known-hosts file required')
if any(not re.fullmatch('SHA256:[A-Za-z0-9+/]{43}=?',s.get('host_key_sha256','')) for s in config['hosts'].values()):raise ValueError('Approved explicit host fingerprints required')
self.factory = client_factory
self.local = threading.local()
self.clients = []
self.lock = threading.Lock()
def get(self, rid):
sessions = getattr(self.local, 'sessions', None)
if sessions is None:
sessions = self.local.sessions = {}
if rid in sessions:
c = sessions[rid]
if c.get_transport() and c.get_transport().is_active():
return c
# Do not transparently retry operations after a lost connection.
raise RuntimeError('SSH session lost; probe remote outcome on next run')
spec = self.config['hosts'][rid]
jump = self.get(self.jump_id) if rid != self.jump_id else None
if self.factory:
client = self.factory(rid, spec, jump)
else:
import paramiko
client = paramiko.SSHClient()
known=Path(self.config['known_hosts'])
if not known.is_absolute() or known.is_symlink() or not known.is_file() or known.stat().st_mode & 0o022:raise ValueError('Trusted non-writable known-hosts reference required')
entries=paramiko.HostKeys(str(known))
hostname=spec['host'] if spec.get('port',22)==22 else '['+spec['host']+']:'+str(spec['port'])
approved=[]
for key in (entries.lookup(hostname) or {}).values():
actual='SHA256:'+base64.b64encode(hashlib.sha256(key.asbytes()).digest()).decode().rstrip('=')
if actual==spec['host_key_sha256'].rstrip('='):approved.append(key)
if not approved:raise ValueError('Approved known-host key absent; no credentials transmitted')
for key in approved:client.get_host_keys().add(hostname,key.get_name(),key)
client.set_missing_host_key_policy(paramiko.RejectPolicy())
auth = credentials(spec)
if set(auth) - {'username', 'password', 'key_filename', 'passphrase'}:
raise ValueError('Unexpected SSH credential fields')
if auth.get('username',spec['username'])!=spec['username']:raise ValueError('Credential username differs from registered host')
if auth.get('key_filename'):
key=Path(auth['key_filename'])
if not key.is_absolute() or not key.is_file() or key.stat().st_mode & 0o077:raise ValueError('Private absolute SSH key required')
sock = jump.get_transport().open_channel('direct-tcpip', (spec['host'], spec.get('port', 22)), ('127.0.0.1', 0)) if jump else None
try:
client.connect(spec['host'], port=spec.get('port', 22), username=spec['username'],
password=auth.get('password'), key_filename=auth.get('key_filename'),
passphrase=auth.get('passphrase'), sock=sock, allow_agent=False,
look_for_keys=False, timeout=20, banner_timeout=20, auth_timeout=20)
except Exception:
client.close()
raise RuntimeError('SSH connect failed; inspect registered route/host key/credential reference') from None
finally:
auth.clear()
sessions[rid] = client
with self.lock:
self.clients.append(client)
return client
def close(self):
for client in reversed(self.clients):
client.close()
self.clients.clear()