Tools/Volatility

#4 Volatility - URL 추출 플러그인 만들기

geunyeong 2021. 12. 18. 01:43

Table of Contents

    Abstract

    이번 글에서는 프로세스의 가상 메모리에서 URL 문자열을 추출하는 플러그인을 만들어 본다. 플러그인 명령행 인자로 PID나 프로세스 이름을 선택적으로 입력받는다. 둘 다 입력되지 않으면 모든 프로세스의 가상 메모리에서 URL 문자열을 추출하고, PID가 주어지면 해당 프로세스만, 프로세스 이름이 주어지면 해당 프로세스들에서 URL 문자열을 추출한다.

    Volatility3 Documentation의 심플 플러그인 만드는 법을 따라했던 전 글을 참고하면 좋을 것 같다.

    • #3 Volatility - 간단 플러그인 만들기
    https://geun-yeong.tistory.com/61

    Make an URL Finder Plugin

    Declare the Plugin Class

    플러그인 클래스 먼저 선언한다. 앞선 글들에서도 보았듯 PluginInterface를 상속받은 클래스를 생성한다.

    from volatility3.framework import interfaces
    
    class MyUrlFinder(interfaces.plugins.PluginInterface):
        """My URL Finder Plugin"""
    
        _required_framework_version = (1, 0, 0)
        _version = (1, 0, 0)

    Define Requirements

    pslist.PsList 클래스의 기능을 사용하기 위한 TranslationLayerRequirement와 SymbolTableRequirement를 추가하고, 주인공인 pslist를 PluginRequirement를 통해 requirement로 등록한다. 또한 명령행에서 플러그인의 인자로 받을 PID와 Process Name을 위해 IntRequirement와 StringRequirement도 추가했다. IntRequirement는 pid라는 이름으로 MyUrlFinder 클래스의 config 속성으로 들어오고, StringRequirement는 pname이라는 이름으로 config에 저장된다. optional을 True로 설정해 둘 다 필수 인자가 아님을 명시하도록 했다.

    from volatility3.framework import renderers
    from volatility3.framework.configuration import requirements
    from volatility3.plugins.windows import pslist
    
    class MyUrlFinder(interfaces.plugins.PluginInterface):
    
    # ...(중략)...
    
        @classmethod
        def get_requirements(self):
            return [
                requirements.TranslationLayerRequirement(
                    name='primary',
                    description='Memory layer for the kernel',
                    architectures=['Intel32', 'Intel64']
                ),
                requirements.SymbolTableRequirement(
                    name='nt_symbols',
                    description='Windows kernel symbols'
                ),
                requirements.PluginRequirement(
                    name='pslist',
                    plugin=pslist.PsList,
                    version=(2, 0, 0)
                ),
                requirements.IntRequirement(
                    name='pid',
                    description='Process ID to include (all other processes are excluded)',
                    optional=True
                ),
                requirements.StringRequirement(
                    name='pname',
                    description='Process name to include (all other processes are excluded)',
                    optional=True
                )
            ]

    Write a run method

    Volatility3 Framework가 호출한 run 메소드를 작성한다. TreeGrid로 표현될 테이블의 타이틀 이름과 데이터 타입을 명시하고, 테이블 내용을 채워줄 _generator 메소드를 호출한다.

    no는 라인 수를 출력하기 위해 추가한 컬럼이고, pid는 프로세스 ID, pname은 프로세스 이름, url은 해당 프로세스의 가상 메모리에서 추출한 URL 문자열이다.

    pslist.PsList의 create_pid_filter는 PID가 담긴 list를 받는데, IntRequirement는 정수 하나만을 받아 config에 넣어주기 때문에 create_pid_filter에 list로 변환해 넘겨줘야 한다.

    from volatility3.framework import renderers
    from volatility3.plugins.windows import pslist
    
    class MyUrlFinder(interfaces.plugins.PluginInterface):
    
    #...(중략)...
    
        def run(self):
            return renderers.TreeGrid(
                [
                    # colums name and type
                    ('no',    int),
                    ('pid',   int),
                    ('pname', str),
                    ('url',   str)
                ],
                self._generator(
                    pslist.PsList.list_processes(
                        self.context,
                        self.config['primary'],
                        self.config['nt_symbols'],
                        filter_func = pslist.PsList.create_pid_filter(
                            [self.config.get('pid', None)]
                        )
                    ),
                    self.config.get('pname', None)
                )
            )

    Write a _generator method

    BLOB 형태인 가상 메모리 덤프에서 URL을 추출하기 위해 정규표현식을 사용했다. 플러그인 클래스 내부에 정규표현식 객체(_strings_regex, _url_regex)를 생성하고, _generator 메소드에서 이를 사용해 문자열과 URL을 추출하도록 했다.

    pslist.PsList로 받은 프로세스 객체의 get_vad_root 메소드를 사용해 프로세스의 가상 메모리 관련 데이터가 저장된 VAD 리스트를 받을 수 있다. 이를 for문으로 하나씩 돌아가며 VAD 덤프 데이터를 바이트 배열로 가져온 후 정규표현식으로 문자열만 먼저 추출한다(바이너리인 상태에서 곧바로 URL을 추출하려니 생각보다 힘들더라). 추출된 문자열에서 또다시 URL을 추출하는 정규표현식을 돌려 URL을 뽑아내고 이를 yield로 리스트에 추가하도록 했다.

    import re
    
    class MyUrlFinder(interfaces.plugins.PluginInterface):
    
        _strings_regex = re.compile(b'[\x20-\x7E]+')
        _url_regex = re.compile(b'https?\:\/\/[a-zA-Z0-9\.\/\?\:@\-_=#]+\.[a-zA-Z]{2,6}[a-zA-Z0-9\.\&\/\?\:@\-_=#]*')
    
    #...(중략)...
    
        def _generator(self, procs, pname):
            num = 0
    
            for proc in procs:
                proc_name = proc.ImageFileName.cast(
                    'string',
                    max_length=proc.ImageFileName.vol.count,
                    errors='replace'
                )
                
                if pname and not pname.lower().startswith(proc_name.lower()):
                    continue
    
                for vad in proc.get_vad_root().traverse():
                    try:
                        proc_layer_name = proc.add_process_layer()
                        proc_layer = self.context.layers[proc_layer_name]
    
                        data_size = vad.get_end() - vad.get_start()
                        data = proc_layer.read(vad.get_start(), data_size, pad=True)
                        
                        for string in self._strings_regex.findall(data):
                            for url in self._url_regex.findall(string):
                                yield (
                                    0,                        # level
                                    (
                                        num,                  # no
                                        proc.UniqueProcessId, # pid
                                        proc_name,            # pname
                                        url.decode()          # url
                                    )
                                )
                                num += 1
                    except MemoryError:
                        pass

    Run

    URL Finder 플러그인 테스트는 Otter CTF에 출제되었던 메모리 포렌식 문제 덤프 파일을 사용했다. 메모리 덤프의 프로필은 Windows 7 SP1이다.

    With a PID

    --pid 옵션을 사용해 Otter CTF 메모리 덤프에서 PID가 708인 프로세스의 가상메모리에서 URL을 추출했다. 8개 밖에 나오지 않았는데, 분명 이보다 더 많을 것 같다. 아마 유니코드로 된 URL은 출력되지 않은 게 아닐까 싶지만 일단 URL을 잘 추출하는 걸 볼 수 있다.

    PID가 708인 프로세스에서 추출한 URL

    With a Process Name

    --pname 옵션을 사용해 프로세스 이름이 LunarMS.exe인 프로세스의 가상 메모리에서 URL을 추출하도록 했다. 편의를 위해 대소문자를 구분하지 않도록 해서 소문자로 입력했음에도 LunarMS.exe 프로세스에서 추출한 URL이 나오고, 위의 PID로 추출한 URL 결과와 동일한 것을 볼 수 있다.

    프로세스 이름이 LunarMS.exe인 프로세스에서 추출한 URL

    About all processes

    아무런 옵션을 주지 않고 URL Finder 플러그인을 실행하면 모든 프로세스의 가상 메모리에서 URL을 추출하며, 총 27000여개에 달하는 URL이 출력된다.

    전체 프로세스에서 추출한 모든 URL

    Close

    이상으로 각 프로세스의 가상 메모리에 접근해 가상 메모리 덤프 데이터를 가져오고 여기서 우리가 원하는 정보를 추출하는 방법을 알아봤다. 여기까지 했다면 대부분 자신이 원하는 플러그인을 어느정도 만들 줄 알게 되지 않을까 생각한다. 본 URL Finder 플러그인 전체 코드는 아래 Source Code 절에 작성했다. 

    Source Code

    더보기
    #!/usr/bin/env python
    # -*- coding: utf-8 -*-
    
    __author__  = "Geunyeong Choi"
    __email__   = "lucete387@gmail.com"
    __version__ = "1.0.0"
    
    from volatility3.framework import interfaces, renderers
    from volatility3.framework.configuration import requirements
    from volatility3.plugins.windows import pslist
    
    import re
    
    class MyUrlFinder(interfaces.plugins.PluginInterface):
        """My URL Finder Plugin"""
    
        _required_framework_version = (1, 0, 0)
        _version = (1, 0, 0)
    
        _strings_regex = re.compile(b'[\x20-\x7E]+')
        _url_regex = re.compile(b'https?\:\/\/[a-zA-Z0-9\.\/\?\:@\-_=#]+\.[a-zA-Z]{2,6}[a-zA-Z0-9\.\&\/\?\:@\-_=#]*')
    
        @classmethod
        def get_requirements(self):
            return [
                requirements.TranslationLayerRequirement(
                    name='primary',
                    description='Memory layer for the kernel',
                    architectures=['Intel32', 'Intel64']
                ),
                requirements.SymbolTableRequirement(
                    name='nt_symbols',
                    description='Windows kernel symbols'
                ),
                requirements.PluginRequirement(
                    name='pslist',
                    plugin=pslist.PsList,
                    version=(2, 0, 0)
                ),
                requirements.IntRequirement(
                    name='pid',
                    description='Process ID to include (all other processes are excluded)',
                    optional=True
                ),
                requirements.StringRequirement(
                    name='pname',
                    description='Process name to include (all other processes are excluded)',
                    optional=True
                )
            ]
    
        def _generator(self, procs, pname):
            num = 0
    
            for proc in procs:
                proc_name = proc.ImageFileName.cast('string',
                                                    max_length=proc.ImageFileName.vol.count,
                                                    errors='replace')
                
                if pname and not pname.lower().startswith(proc_name.lower()):
                    continue
    
                for vad in proc.get_vad_root().traverse():
                    try:
                        proc_layer_name = proc.add_process_layer()
                        proc_layer = self.context.layers[proc_layer_name]
    
                        data_size = vad.get_end() - vad.get_start()
                        data = proc_layer.read(vad.get_start(), data_size, pad=True)
                        
                        for string in self._strings_regex.findall(data):
                            for url in self._url_regex.findall(string):
                                yield (
                                    0,                        # level
                                    (
                                        num,                  # no
                                        proc.UniqueProcessId, # pid
                                        proc_name,            # pname
                                        url.decode()          # url
                                    )
                                )
                                num += 1
                    except MemoryError:
                        comment  =  """
                                    yield (
                                        0,
                                        (
                                            -1,
                                            proc.UniqueProcessId, 
                                            proc_name,
                                            '<Too large memory (pid: {}, size: 0x{:x} bytes)>'.format(proc.UniqueProcessId, data_size)
                                        )
                                    )
                                    """
                        pass
    
        def run(self):
            return renderers.TreeGrid(
                [
                    # colums name and type
                    ('no',    int),
                    ('pid',   int),
                    ('pname', str),
                    ('url',   str)
                ],
                self._generator(
                    pslist.PsList.list_processes(
                        self.context,
                        self.config['primary'],
                        self.config['nt_symbols'],
                        filter_func = pslist.PsList.create_pid_filter(
                            [self.config.get('pid', None)]
                        )
                    ),
                    self.config.get('pname', None)
                )
            )