주정통 점검 코드.txt
17.84 KB · text/x-script.python · 업로드 2026-07-20 16:20
import paramiko
import time
class MariaDBVulnerabilityScanner:
def __init__(self, host, port, username, password=None, key_filename=None, db_user="root", db_pass=None):
"""
SSH 접속 및 DB 인증 정보 초기화
"""
self.host = host
self.port = port
self.username = username
self.password = password
self.key_filename = key_filename
# DB 쿼리용 계정 (DB 패스워드가 없거나 unix_socket을 사용한다면 db_pass=None 유지)
self.db_user = db_user
self.db_pass = db_pass
self.ssh = paramiko.SSHClient()
self.ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
def connect(self):
"""SSH 연결"""
try:
self.ssh.connect(
hostname=self.host,
port=self.port,
username=self.username,
password=self.password,
key_filename=self.key_filename,
timeout=10
)
print(f"[*] SSH Connection Established to {self.host}")
except Exception as e:
print(f"[!] SSH Connection Failed: {e}")
raise
def disconnect(self):
"""SSH 연결 해제"""
self.ssh.close()
print(f"[*] SSH Connection Closed")
def _execute_cmd(self, cmd):
"""원격 명령어 실행 내부 메서드"""
stdin, stdout, stderr = self.ssh.exec_command(cmd)
# 명령어 실행 완료 대기
exit_status = stdout.channel.recv_exit_status()
out = stdout.read().decode('utf-8').strip()
err = stderr.read().decode('utf-8').strip()
return exit_status, out, err
def _execute_sql(self, query):
"""
DB SQL 쿼리 실행 내부 메서드 (CLI 기반)
Ubuntu/Debian 계열의 경우 root 계정은 sudo mysql로 패스워드 없이 소켓 접속이 가능함
"""
if self.db_pass:
cmd = f'mysql -u{self.db_user} -p"{self.db_pass}" -e "{query}"'
else:
# unix_socket 인증을 활용한 sudo 접속 (패스워드 노출 방지)
cmd = f'sudo mysql -e "{query}"'
return self._execute_cmd(cmd)
# ==========================================
# 1. OS 레벨 점검 함수 (CLI)
# ==========================================
def check_d07_process_owner(self):
"""D-07: root 권한으로 서비스 구동 제한"""
print("\n[+] D-07 점검: 프로세스 구동 계정 확인")
cmd = "ps -ef | grep mysqld | grep -v grep | awk '{print $1}'"
status, out, err = self._execute_cmd(cmd)
if out:
print(f" -> 실행 계정:\n{out}")
if "root" in out.split('\n'):
print(" -> [취약] root 권한으로 DB 서비스가 구동 중입니다.")
else:
print(" -> [양호] 안전한 계정으로 구동 중입니다.")
else:
print(" -> DB 프로세스를 찾을 수 없습니다.")
def check_d10_network_binding(self):
"""D-10: 원격에서 DB 서버로의 접속 제한"""
print("\n[+] D-10 점검: 네트워크 바인딩 상태 확인")
cmd = "ss -tulnp | grep ':3306' || netstat -tulnp | grep ':3306'"
status, out, err = self._execute_cmd(cmd)
if out:
print(f" -> 포트 상태:\n{out}")
if "127.0.0.1" in out:
print(" -> [양호] 로컬 호스트(127.0.0.1)에만 바인딩 되어 있습니다.")
else:
print(" -> [취약/주의] 외부 접속이 허용된 상태(0.0.0.0 등)입니다. 방화벽 확인이 필요합니다.")
else:
print(" -> 3306 포트 리스닝 정보를 찾을 수 없습니다.")
def check_d14_config_permissions(self):
"""D-14: 주요 설정파일 권한 적절성"""
print("\n[+] D-14 점검: 주요 설정 파일 및 디렉터리 권한 확인")
# 주요 경로 (Ubuntu/Debian 기준)
cmd = "ls -ld /etc/mysql /var/lib/mysql /etc/mysql/my.cnf 2>/dev/null"
status, out, err = self._execute_cmd(cmd)
if out:
print(f" -> 파일/디렉터리 권한:\n{out}")
# 권한 파싱 로직은 추가 분석 필요 (일단 출력만 제공)
else:
print(" -> 지정된 경로의 파일/디렉터리를 찾을 수 없습니다.")
def check_d25_version(self):
"""D-25: DB 버전 및 패치 확인 (dpkg 활용)"""
print("\n[+] D-25 점검: DB 버전 정보")
cmd = "dpkg -l | grep -E 'mariadb-server|mysql-server' || mysql -V"
status, out, err = self._execute_cmd(cmd)
print(f" -> 패키지/버전 정보:\n{out if out else err}")
# ==========================================
# 2. DB 레벨 점검 함수 (SQL 쿼리 CLI 전송)
# ==========================================
def check_d01_d02_accounts(self):
"""D-01/D-02: 기본/불필요 계정 상태 점검"""
print("\n[+] D-01/D-02 점검: DB 사용자 목록 및 호스트 조회")
query = "SELECT User, Host, plugin FROM mysql.user;"
status, out, err = self._execute_sql(query)
if not err:
print(f" -> 계정 목록:\n{out}")
else:
print(f" -> [오류] 쿼리 실행 실패: {err}")
def check_d04_dba_privileges(self):
"""D-04: 데이터베이스 관리자 권한 제한"""
print("\n[+] D-04 점검: Super_priv 권한을 가진 계정 조회")
query = "SELECT User, Host FROM mysql.user WHERE Super_priv = 'Y';"
status, out, err = self._execute_sql(query)
if not err:
print(f" -> 최고 관리자 권한 보유 계정:\n{out}")
else:
print(f" -> [오류] 쿼리 실행 실패: {err}")
def check_d21_grant_option(self):
"""D-21: 인가되지 않은 GRANT OPTION 제한"""
print("\n[+] D-21 점검: Grant_priv 권한을 가진 계정 조회")
query = "SELECT User, Host FROM mysql.user WHERE Grant_priv = 'Y';"
status, out, err = self._execute_sql(query)
if not err:
print(f" -> Grant 권한(타인에게 권한 부여) 보유 계정:\n{out}")
else:
print(f" -> [오류] 쿼리 실행 실패: {err}")
def check_d26_audit_plugin(self):
"""D-26: 감사(Audit) 로깅 활성화 여부"""
print("\n[+] D-26 점검: Audit 플러그인 설치 여부")
query = "SHOW PLUGINS;"
status, out, err = self._execute_sql(query)
if not err:
if "audit" in out.lower():
print(" -> [양호] Audit 관련 플러그인이 로드되어 있습니다.")
else:
print(" -> [취약] Audit 플러그인이 확인되지 않습니다.")
else:
print(f" -> [오류] 쿼리 실행 실패: {err}")
def check_d03_password_complexity(self):
"""D-03: 비밀번호의 사용기간 및 복잡도 설정"""
print("\n[+] D-03 점검: 패스워드 복잡도 및 만료 기간 플러그인 확인")
query = "SHOW VARIABLES LIKE '%password%';"
status, out, err = self._execute_sql(query)
if not err:
print(f" -> 패스워드 관련 설정:\n{out}")
if "validate_password" not in out and "password_lifetime" not in out:
print(" -> [취약/주의] 패스워드 복잡도(validate_password) 플러그인이나 만료 기간 설정이 보이지 않을 수 있습니다.")
else:
print(f" -> [오류] 쿼리 실행 실패: {err}")
def check_d05_password_reuse(self):
"""D-05: 비밀번호 재사용 제약"""
print("\n[+] D-05 점검: 패스워드 재사용 제한 설정 확인")
query = "SHOW VARIABLES LIKE 'password_history';"
status, out, err = self._execute_sql(query)
if not err:
print(f" -> 패스워드 재사용 제한(password_history):\n{out}")
else:
print(f" -> [오류] 쿼리 실행 실패: {err}")
def check_d06_individual_accounts(self):
"""D-06: DB 사용자 계정 개별 부여 (수동 점검 필요)"""
print("\n[+] D-06 점검: DB 공용 계정 사용 여부 (시스템적으로 자동 판별 불가)")
print(" -> [수동 점검] 애플리케이션 연동 계정 외에 개발자/관리자가 공용 계정(admin 등)을 공유하는지 인터뷰 및 접속 로그를 통해 확인해야 합니다.")
def check_d08_encryption_algorithm(self):
"""D-08: 안전한 암호화 알고리즘 사용"""
print("\n[+] D-08 점검: 계정별 패스워드 암호화 플러그인 확인")
query = "SELECT User, Host, plugin FROM mysql.user;"
status, out, err = self._execute_sql(query)
if not err:
print(f" -> 플러그인 정보:\n{out}")
if "mysql_old_password" in out:
print(" -> [취약] 구버전의 취약한 해시 알고리즘(mysql_old_password)을 사용하는 계정이 있습니다.")
else:
print(f" -> [오류] 쿼리 실행 실패: {err}")
def check_d09_login_failure_lock(self):
"""D-09: 로그인 실패 시 잠금 정책"""
print("\n[+] D-09 점검: Connection Control 플러그인(로그인 실패 잠금) 확인")
query = "SHOW VARIABLES LIKE 'connection_control%';"
status, out, err = self._execute_sql(query)
if not err:
print(f" -> 실패 잠금 플러그인 설정:\n{out}")
if not out.strip():
print(" -> [취약] Connection Control 플러그인이 설치/활성화되지 않았습니다.")
else:
print(f" -> [오류] 쿼리 실행 실패: {err}")
def check_d11_system_table_access(self):
"""D-11: 인가되지 않은 사용자의 시스템 테이블 접근 제한"""
print("\n[+] D-11 점검: mysql(시스템 DB)에 대한 권한을 가진 일반 계정 확인")
query = "SELECT User, Host, Db, Select_priv FROM mysql.db WHERE Db = 'mysql' AND User NOT IN ('root', 'mysql');"
status, out, err = self._execute_sql(query)
if not err:
if out.strip():
print(f" -> [취약] 시스템 DB(mysql)에 권한이 부여된 일반 계정 발견:\n{out}")
else:
print(" -> [양호] 시스템 DB에 비인가 계정의 권한이 부여되지 않았습니다.")
else:
print(f" -> [오류] 쿼리 실행 실패: {err}")
def check_d12_oracle_listener_pw(self):
"""D-12: 안전한 리스너 비밀번호 설정 (Oracle 전용)"""
print("\n[+] D-12 점검: 리스너 비밀번호 설정 (Oracle 전용 항목 -> MariaDB N/A)")
print(" -> MariaDB/MySQL 환경이므로 점검 생략 (N/A)")
def check_d13_odbc_oledb(self):
"""D-13: 불필요한 ODBC/OLE-DB 데이터 소스 제거 (Linux OS 레벨)"""
print("\n[+] D-13 점검: Linux odbcinst.ini 파일 확인")
cmd = "cat /etc/odbcinst.ini 2>/dev/null"
status, out, err = self._execute_cmd(cmd)
if out:
print(f" -> 등록된 ODBC 드라이버 목록:\n{out}")
else:
print(" -> /etc/odbcinst.ini 파일이 존재하지 않거나 등록된 드라이버가 없습니다. [양호]")
def check_d15_oracle_listener_restrict(self):
"""D-15: 오라클 리스너 로깅/trace 변경 제한 (Oracle 전용)"""
print("\n[+] D-15 점검: 리스너 로그 변경 제한 (Oracle 전용 항목 -> MariaDB N/A)")
print(" -> MariaDB/MySQL 환경이므로 점검 생략 (N/A)")
def check_d16_windows_auth(self):
"""D-16: Windows 인증 모드 사용 (MSSQL 전용)"""
print("\n[+] D-16 점검: Windows 인증 모드 (MSSQL 전용 항목 -> MariaDB N/A)")
print(" -> MariaDB/MySQL 환경이므로 점검 생략 (N/A)")
def check_d17_audit_table_access(self):
"""D-17: Audit Table 접근 제한"""
print("\n[+] D-17 점검: 일반 로그/감사 로그 테이블 접근 권한 확인")
# MariaDB의 general_log 등의 접근 권한을 확인
query = "SELECT User, Host, Db, Table_name FROM mysql.tables_priv WHERE Table_name LIKE '%log%';"
status, out, err = self._execute_sql(query)
if not err:
print(f" -> 로그 관련 테이블 권한 보유 계정:\n{out if out.strip() else '특이사항 없음 (양호)'}")
else:
print(f" -> [오류] 쿼리 실행 실패: {err}")
def check_d18_public_role(self):
"""D-18: Role이 Public으로 설정되지 않도록 조정"""
print("\n[+] D-18 점검: Anonymous(익명) 계정 존재 여부 및 PUBLIC 롤 확인")
query = "SELECT User, Host FROM mysql.user WHERE User = '';"
status, out, err = self._execute_sql(query)
if not err:
if out.strip():
print(f" -> [취약] 익명(Anonymous) 계정이 존재합니다:\n{out}")
else:
print(" -> [양호] 익명 계정이 존재하지 않습니다.")
else:
print(f" -> [오류] 쿼리 실행 실패: {err}")
def check_d19_remote_os_auth(self):
"""D-19: 원격 OS 인증 파라미터 FALSE 설정 (Oracle 전용)"""
print("\n[+] D-19 점검: REMOTE_OS_AUTHENTICATION (Oracle 전용 항목 -> MariaDB N/A)")
print(" -> MariaDB/MySQL 환경이므로 점검 생략 (N/A)")
def check_d20_unauthorized_object_owner(self):
"""D-20: 인가되지 않은 Object owner의 제한"""
print("\n[+] D-20 점검: FEDERATED 엔진(DB Link 유사) 사용 여부 확인")
query = "SHOW ENGINES;"
status, out, err = self._execute_sql(query)
if not err:
if "FEDERATED" in out and "YES" in out.split("FEDERATED")[1].split()[0]:
print(" -> [주의] FEDERATED 엔진이 활성화되어 있습니다. 외부 연결 관리가 필요합니다.")
else:
print(" -> [양호] FEDERATED 엔진이 비활성화되어 있거나 지원하지 않습니다.")
else:
print(f" -> [오류] 쿼리 실행 실패: {err}")
def check_d22_resource_limit(self):
"""D-22: 데이터베이스 자원 제한 기능"""
print("\n[+] D-22 점검: 계정별 리소스 제한 (max_user_connections 등) 확인")
query = "SELECT User, Host, max_questions, max_updates, max_connections, max_user_connections FROM mysql.user;"
status, out, err = self._execute_sql(query)
if not err:
print(f" -> 사용자별 리소스 제한 설정:\n{out}")
else:
print(f" -> [오류] 쿼리 실행 실패: {err}")
def check_d23_xp_cmdshell(self):
"""D-23: xp_cmdshell 사용 제한 (MSSQL 항목 -> MariaDB FILE 권한 점검으로 대체)"""
print("\n[+] D-23 점검: FILE 권한(OS 명령 실행 우회 가능) 부여 계정 확인 (MariaDB 대체 점검)")
query = "SELECT User, Host FROM mysql.user WHERE File_priv = 'Y';"
status, out, err = self._execute_sql(query)
if not err:
print(f" -> FILE 권한(파일 읽기/쓰기 가능) 보유 계정:\n{out}")
else:
print(f" -> [오류] 쿼리 실행 실패: {err}")
def check_d24_registry_procedure(self):
"""D-24: Registry Procedure 권한 제한 (MSSQL 전용)"""
print("\n[+] D-24 점검: Registry Procedure 제한 (MSSQL 전용 항목 -> MariaDB N/A)")
print(" -> MariaDB/MySQL 환경이므로 점검 생략 (N/A)")
# ==========================================
# 실행 예시 (Main)
# ==========================================
if __name__ == "__main__":
# 서버 접속 정보 세팅 (테스트 환경에 맞게 수정)
TARGET_HOST = "192.168.1.100"
TARGET_PORT = 22
SSH_USER = "ubuntu" # SSH 접속 계정
SSH_PASS = "your_ssh_password" # 필요시 키 파일(key_filename) 사용 권장
# Scanner 인스턴스 생성
scanner = MariaDBVulnerabilityScanner(
host=TARGET_HOST,
port=TARGET_PORT,
username=SSH_USER,
password=SSH_PASS,
db_user="root",
db_pass=None # sudo mysql로 접근 시 None
)
try:
scanner.connect()
# 1. OS 환경 점검
scanner.check_d07_process_owner()
scanner.check_d10_network_binding()
scanner.check_d14_config_permissions()
scanner.check_d25_version()
# 2. DB 내부 환경 점검 (CLI 통해 쿼리 실행)
scanner.check_d01_d02_accounts()
scanner.check_d04_dba_privileges()
scanner.check_d21_grant_option()
scanner.check_d26_audit_plugin()
scanner.check_d03_password_complexity()
scanner.check_d05_password_reuse()
scanner.check_d06_individual_accounts()
scanner.check_d08_encryption_algorithm()
scanner.check_d09_login_failure_lock()
scanner.check_d11_system_table_access()
scanner.check_d12_oracle_listener_pw()
scanner.check_d13_odbc_oledb()
scanner.check_d15_oracle_listener_restrict()
scanner.check_d16_windows_auth()
scanner.check_d17_audit_table_access()
scanner.check_d18_public_role()
scanner.check_d19_remote_os_auth()
scanner.check_d20_unauthorized_object_owner()
scanner.check_d22_resource_limit()
scanner.check_d23_xp_cmdshell()
scanner.check_d24_registry_procedure()
except Exception as e:
print(f"점검 중 오류 발생: {e}")
finally:
scanner.disconnect()