proxmox-rdp-autosetup/rdp_control_web.py

#!/usr/bin/env python3

import os
import subprocess
import json
from flask import Flask, render_template, jsonify, request
from flask_cors import CORS
import logging
from pathlib import Path

app = Flask(__name__)
CORS(app)

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

CONFIG_FILE = "/etc/xrdp/auto_login.conf"
SYSTEMD_SERVICE = "xrdp-autologin.service"

def get_auto_login_status():
    """자동 로그인 상태 확인"""
    try:
        # systemd 서비스 상태 확인
        result = subprocess.run(
            ["systemctl", "is-active", SYSTEMD_SERVICE],
            capture_output=True,
            text=True
        )
        service_active = result.stdout.strip() == "active"
        
        # 설정 파일 존재 여부 확인
        config_exists = os.path.exists(CONFIG_FILE)
        
        # systemd 서비스 enabled 상태 확인
        result = subprocess.run(
            ["systemctl", "is-enabled", SYSTEMD_SERVICE],
            capture_output=True,
            text=True
        )
        service_enabled = result.stdout.strip() == "enabled"
        
        return {
            "enabled": service_enabled,
            "active": service_active,
            "config_exists": config_exists
        }
    except Exception as e:
        logger.error(f"상태 확인 실패: {str(e)}")
        return {
            "enabled": False,
            "active": False,
            "config_exists": False,
            "error": str(e)
        }

def set_auto_login(enabled):
    """자동 로그인 활성화/비활성화"""
    try:
        if enabled:
            # 서비스 활성화 및 시작
            subprocess.run(["systemctl", "enable", SYSTEMD_SERVICE], check=True)
            subprocess.run(["systemctl", "start", SYSTEMD_SERVICE], check=True)
            
            # xrdp 서비스 재시작
            subprocess.run(["systemctl", "restart", "xrdp"], check=True)
            subprocess.run(["systemctl", "restart", "xrdp-sesman"], check=True)
            
            logger.info("RDP 자동 로그인 활성화됨")
            return True, "RDP 자동 로그인이 활성화되었습니다."
        else:
            # 서비스 중지 및 비활성화
            subprocess.run(["systemctl", "stop", SYSTEMD_SERVICE], check=True)
            subprocess.run(["systemctl", "disable", SYSTEMD_SERVICE], check=True)
            
            # 자동 로그인 설정 제거
            if os.path.exists(CONFIG_FILE):
                backup_file = f"{CONFIG_FILE}.backup"
                subprocess.run(["mv", CONFIG_FILE, backup_file], check=True)
            
            # xrdp 서비스 재시작
            subprocess.run(["systemctl", "restart", "xrdp"], check=True)
            subprocess.run(["systemctl", "restart", "xrdp-sesman"], check=True)
            
            logger.info("RDP 자동 로그인 비활성화됨")
            return True, "RDP 자동 로그인이 비활성화되었습니다."
    except subprocess.CalledProcessError as e:
        logger.error(f"명령 실행 실패: {str(e)}")
        return False, f"설정 변경 실패: {str(e)}"
    except Exception as e:
        logger.error(f"예외 발생: {str(e)}")
        return False, f"오류 발생: {str(e)}"

def get_system_info():
    """시스템 정보 가져오기"""
    try:
        # 호스트명
        hostname = subprocess.run(
            ["hostname"],
            capture_output=True,
            text=True
        ).stdout.strip()
        
        # IP 주소
        ip_result = subprocess.run(
            ["ip", "-4", "addr", "show", "scope", "global"],
            capture_output=True,
            text=True
        )
        ip_addresses = []
        for line in ip_result.stdout.split('\n'):
            if 'inet' in line:
                ip = line.strip().split()[1].split('/')[0]
                ip_addresses.append(ip)
        
        # RDP 포트 확인
        rdp_port = "3389"
        
        # Tailscale 상태
        tailscale_ip = None
        try:
            ts_result = subprocess.run(
                ["tailscale", "ip", "-4"],
                capture_output=True,
                text=True
            )
            if ts_result.returncode == 0:
                tailscale_ip = ts_result.stdout.strip()
        except:
            pass
        
        return {
            "hostname": hostname,
            "ip_addresses": ip_addresses,
            "rdp_port": rdp_port,
            "tailscale_ip": tailscale_ip
        }
    except Exception as e:
        logger.error(f"시스템 정보 가져오기 실패: {str(e)}")
        return {}

@app.route('/')
def index():
    """메인 페이지"""
    return render_template('index.html')

@app.route('/api/status', methods=['GET'])
def get_status():
    """현재 상태 조회 API"""
    status = get_auto_login_status()
    system_info = get_system_info()
    return jsonify({
        "status": status,
        "system": system_info
    })

@app.route('/api/toggle', methods=['POST'])
def toggle_auto_login():
    """자동 로그인 토글 API"""
    data = request.get_json()
    enabled = data.get('enabled', False)
    
    success, message = set_auto_login(enabled)
    
    if success:
        return jsonify({
            "success": True,
            "message": message,
            "status": get_auto_login_status()
        })
    else:
        return jsonify({
            "success": False,
            "message": message,
            "status": get_auto_login_status()
        }), 500

@app.route('/api/logs', methods=['GET'])
def get_logs():
    """최근 로그 조회 API"""
    try:
        # xrdp 로그
        xrdp_logs = subprocess.run(
            ["journalctl", "-u", "xrdp", "-n", "20", "--no-pager"],
            capture_output=True,
            text=True
        ).stdout
        
        # 자동 로그인 서비스 로그
        autologin_logs = subprocess.run(
            ["journalctl", "-u", SYSTEMD_SERVICE, "-n", "20", "--no-pager"],
            capture_output=True,
            text=True
        ).stdout
        
        return jsonify({
            "xrdp_logs": xrdp_logs,
            "autologin_logs": autologin_logs
        })
    except Exception as e:
        return jsonify({"error": str(e)}), 500

@app.route('/api/restart', methods=['POST'])
def restart_services():
    """서비스 재시작 API"""
    try:
        subprocess.run(["systemctl", "restart", "xrdp"], check=True)
        subprocess.run(["systemctl", "restart", "xrdp-sesman"], check=True)
        
        status = get_auto_login_status()
        if status['enabled']:
            subprocess.run(["systemctl", "restart", SYSTEMD_SERVICE], check=True)
        
        return jsonify({
            "success": True,
            "message": "서비스가 재시작되었습니다."
        })
    except Exception as e:
        return jsonify({
            "success": False,
            "message": f"재시작 실패: {str(e)}"
        }), 500

if __name__ == '__main__':
    # templates 디렉토리 생성
    Path("templates").mkdir(exist_ok=True)
    
    # 개발 서버 실행 (프로덕션에서는 gunicorn 사용 권장)
    app.run(host='0.0.0.0', port=5000, debug=False)