207 lines
6.2 KiB
Python
207 lines
6.2 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
端口占用检测器 v1.0
|
||
功能：
|
||
1. 先UAC提权到管理员
|
||
2. 检测用户输入指定端口是否被占用
|
||
3. 如果被占用，找出是哪个进程占用
|
||
4. 允许用户强制杀死该进程
|
||
5. 支持批量查询（逗号/空格/范围分隔）
|
||
"""
|
||
import os
|
||
import sys
|
||
import ctypes
|
||
import subprocess
|
||
import re
|
||
|
||
|
||
# ============ UAC 提权 ============
|
||
def is_admin():
|
||
try:
|
||
return ctypes.windll.shell32.IsUserAnAdmin() != 0
|
||
except Exception:
|
||
return False
|
||
|
||
|
||
def request_admin():
|
||
"""请求 UAC 提权，以管理员身份重新运行"""
|
||
if not is_admin():
|
||
print("[!] 需要管理员权限，正在请求 UAC 提权...")
|
||
# 重新以管理员运行自身
|
||
params = ' '.join([f'"{a}"' for a in sys.argv])
|
||
ctypes.windll.shell32.ShellExecuteW(
|
||
None, "runas", sys.executable, f'"{sys.argv[0]}" {params}', None, 1)
|
||
sys.exit(0)
|
||
|
||
|
||
OK = '\033[1;32m'
|
||
WARN = '\033[1;33m'
|
||
RED = '\033[1;31m'
|
||
HDR = '\033[1;36m'
|
||
END = '\033[0m'
|
||
|
||
|
||
def parse_input(s):
|
||
"""解析端口输入：支持 '80, 443, 8000-8010, 22' 等格式"""
|
||
ports = set()
|
||
s = s.replace('，', ',')
|
||
for part in s.replace(',', ' ').split():
|
||
part = part.strip()
|
||
if not part:
|
||
continue
|
||
if '-' in part: # 范围
|
||
a, b = part.split('-')
|
||
try:
|
||
a, b = int(a), int(b)
|
||
for p in range(a, b + 1):
|
||
ports.add(p)
|
||
except ValueError:
|
||
continue
|
||
else: # 单个端口
|
||
try:
|
||
ports.add(int(part))
|
||
except ValueError:
|
||
pass
|
||
return sorted(ports)
|
||
|
||
|
||
def get_netstat():
|
||
"""获取所有 TCP 监听/连接状态的端口→PID 映射"""
|
||
# 使用 netstat 获取
|
||
result = subprocess.run(
|
||
['netstat', '-ano'], capture_output=True, text=True, errors='ignore')
|
||
port_pid = {} # 端口 -> (状态, pid)
|
||
for line in result.stdout.splitlines():
|
||
line = line.strip()
|
||
# 匹配形如 TCP 0.0.0.0:80 0.0.0.0:0 LISTENING 1234
|
||
m = re.match(r'^\s*(TCP|UDP)\s+(\S+):(\d+)\s+(\S+)\s+(\w+)\s*(\d*)\s*$', line)
|
||
if m:
|
||
proto, local, port, remote, state, pid = m.groups()
|
||
try:
|
||
port = int(port)
|
||
except ValueError:
|
||
continue
|
||
# 只关心监听和被占用状态
|
||
if 'LISTEN' in state.upper() or (remote and remote.split(':')[-1] != '0'):
|
||
port_pid[port] = (state, pid)
|
||
return port_pid
|
||
|
||
|
||
def get_process_name(pid):
|
||
"""通过 tasklist 获取进程名"""
|
||
try:
|
||
result = subprocess.run(
|
||
['tasklist', '/FI', f'PID eq {pid}'], capture_output=True, text=True, errors='ignore')
|
||
for line in result.stdout.splitlines():
|
||
if pid in line and '.exe' in line.lower():
|
||
return line.split()[0]
|
||
except Exception:
|
||
pass
|
||
return '未知'
|
||
|
||
|
||
def get_process_detail(pid):
|
||
"""获取进程详细信息（命令行等）"""
|
||
try:
|
||
result = subprocess.run(
|
||
['wmic', 'process', 'where', f'ProcessId={pid}', 'get', 'Name,CommandLine', '/format:list'],
|
||
capture_output=True, text=True, errors='ignore')
|
||
name = cmd = ''
|
||
for line in result.stdout.splitlines():
|
||
if line.startswith('Name='):
|
||
name = line.split('=', 1)[1].strip()
|
||
elif line.startswith('CommandLine='):
|
||
cmd = line.split('=', 1)[1].strip()[:120]
|
||
return name or get_process_name(pid), cmd
|
||
except Exception:
|
||
return get_process_name(pid), ''
|
||
|
||
|
||
def kill_process(pid):
|
||
"""强制杀死进程"""
|
||
result = subprocess.run(['taskkill', '/F', '/PID', str(pid)],
|
||
capture_output=True, text=True)
|
||
if result.returncode == 0:
|
||
return True, "已强制终止"
|
||
return False, result.stderr.strip() or "终止失败（可能无权限或进程已退出）"
|
||
|
||
|
||
def check_port(port, port_pid):
|
||
"""检查单个端口"""
|
||
print(f"\n{'─'*60}")
|
||
print(f"{HDR}📡 端口 {port}{END}")
|
||
if port < 0 or port > 65535:
|
||
print(f"{RED}❌ 无效端口范围 (0-65535){END}")
|
||
return
|
||
if port not in port_pid:
|
||
print(f"{OK}✅ 端口 {port} 未被占用{END}")
|
||
return
|
||
|
||
state, pid = port_pid[port]
|
||
pid = pid.strip()
|
||
name = get_process_name(pid)
|
||
cmd = ""
|
||
try:
|
||
n2, cmd = get_process_detail(pid)
|
||
if n2:
|
||
name = n2
|
||
except Exception:
|
||
pass
|
||
|
||
print(f"{RED}⚠️ 端口 {port} 已被占用！{END}")
|
||
print(f" 占用进程: {name}")
|
||
print(f" PID : {pid}")
|
||
print(f" 状态 : {state}")
|
||
if cmd:
|
||
print(f" 命令行 : {cmd}")
|
||
|
||
if name.lower() in ('system', 'idle', 'svchost.exe') :
|
||
print(f"{WARN}⚠️ 这是系统关键进程，不建议强制终止！{END}")
|
||
return
|
||
|
||
# 询问是否杀死
|
||
kill_choice = input(f"\n是否强制终止进程 {name} (PID {pid})? [y/N]: ").strip().lower()
|
||
if kill_choice in ('y', 'yes'):
|
||
ok, msg = kill_process(pid)
|
||
if ok:
|
||
print(f"{OK}✅ {msg}: {name} (PID {pid}){END}")
|
||
else:
|
||
print(f"{RED}❌ {msg}{END}")
|
||
|
||
|
||
def main():
|
||
# ===== UAC 提权 =====
|
||
if os.name == 'nt':
|
||
request_admin()
|
||
|
||
print('=' * 60)
|
||
print(f"{HDR}🔍 端口占用检测器 v1.0{END}")
|
||
print('=' * 60)
|
||
print("支持批量查询: 逗号/空格分隔，支持范围 如 '80, 443, 8000-8010'")
|
||
|
||
while True:
|
||
inp = input("\n请输入要查询的端口 (输入 q 退出): ").strip()
|
||
if inp.lower() in ('q', 'quit', 'exit'):
|
||
break
|
||
if not inp:
|
||
continue
|
||
|
||
ports = parse_input(inp)
|
||
if not ports:
|
||
print(f"{WARN}⚠️ 无法解析端口输入，请检查格式{END}")
|
||
continue
|
||
if len(ports) > 50:
|
||
print(f"{WARN}⚠️ 单次最多查询 50 个端口{END}")
|
||
continue
|
||
|
||
print(f"{OK}⏳ 正在扫描 {len(ports)} 个端口...{END}")
|
||
port_pid = get_netstat()
|
||
for port in ports:
|
||
check_port(port, port_pid)
|
||
|
||
print("\n👋 已退出")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|