codekingpro/portable-devtools

1##########################################################################2#3# pgAdmin 4 - PostgreSQL Tools4#5# Copyright (C) 2013 - 2024, The pgAdmin Development Team6# This software is released under the PostgreSQL License7#8##########################################################################9import logging10import subprocess11from datetime import datetime, timedelta, timezone12from threading import Lock13 14from flask import current_app15 16import config17 18 19class PasswordExec:20 21    lock = Lock()22 23    def __init__(self, cmd, expiration_seconds=None, timeout=60):24        self.cmd = str(cmd)25        self.expiration_seconds = int(expiration_seconds) \26            if expiration_seconds is not None else None27        self.timeout = int(timeout)28        self.password = None29        self.last_result = None30 31    def get(self):32        if config.SERVER_MODE and not config.ENABLE_SERVER_PASS_EXEC_CMD:33            # Arbitrary shell execution on server is a security risk34            raise NotImplementedError('Passexec not available in server mode')35        with self.lock:36            if not self.password or self.is_expired():37                if not self.cmd:38                    return None39                current_app.logger.info('Calling passexec')40                now = datetime.now(timezone.utc)41                try:42                    p = subprocess.run(43                        self.cmd,44                        shell=True,45                        timeout=self.timeout,46                        capture_output=True,47                        text=True,48                        check=True,49                    )50                except subprocess.CalledProcessError as e:51                    if e.stderr:52                        self.create_logger().error(e.stderr)53                    raise54 55                current_app.logger.info('Passexec completed successfully')56                self.last_result = now57                self.password = p.stdout.strip()58            return self.password59 60    def is_expired(self):61        if self.expiration_seconds is None:62            return False63        return self.last_result is not None and\64            datetime.now(timezone.utc) - self.last_result \65            >= timedelta(seconds=self.expiration_seconds)66 67    def create_logger(self):68        logger = logging.getLogger('passexec')69        for h in current_app.logger.handlers:70            logger.addHandler(h)71        return logger72