Python plugin engine architecture (키콤)

목적

1. 동적 모듈 로딩 사용 

• 빌드 후 실행 파일로 존재하는 프로그램의 경우, 
• 내부에서 참조하는 모듈이 수정되면 메인 소스도 다시 컴파일 해서 반영해야 함
• 파일이 큰 경우 재빌드 시간 오래 걸림
• 동적 로딩으로 구현 시 외부 파일 수정하면 런타임에 가져오므로 재빌드 필요 없음.'
• imp 라이브러리 활용

2. 플러그인 구조 필요

• 만약 특정 모듈을 동적으로 로딩해서 사용하는 경우
• 특정 모듈에게 TC가 존재한다면, 모듈 기능이 수정된 후 새로 추가된 TC 뿐만 아니라 기존 TC에 대해서도 전부 테스트 진행 필수
• 일부 기능이 수정되어도 전체 소스에 영향을 미칠 수 있기 때문.
• 예를 들어 악성코드 검색하는 모듈이 검색하는 윈도우 파일 대상 검색 알고리즘을 하나 더 추가한다면 윈도우 파일 뿐만아니라 리눅스, 맥 등등 전체 파일 TC에 대해 동일하게 다시 돌려봐야 함.
• 따라서 윈도우 파일 검색 모듈, 맥 검색 모듈, 리눅스 검색 모듈 분리하여 플러그인 엔진으로 제작 후에 메인 엔진에서 플러그인 앤잔을 가져다 쓸 수있게 하는 것이 좋음.

 

플러그인 구조

• 동일한 이름의 클래스를 가짐
• 클래스 내부 메소드도 동일한 포맷으로 구성 (초기화, 메모리 해제, 검사, 치료, 치료 가능 리스트 반환, 플러그인 정보 반환)
• 커널에서 플러그인 명만 수정해서 사용할 수 있도록 하기 위함
• 플러그인 모듈을 통해 인스턴스 생성 시 , 담당하는 바이러스 명과, 검출 해시 값 등을 저장

 

플러그인 엔진 암호화

• 플러그인을 그냥 배포한다면 사용자가 기능을 수정할 수 있음
• 암호화하여 배포한 후에 사용 시에 복호화 하도록 구현
• 엔진 암호화는 RC4로 암호화하고 RC4키를 RSA를 통해 암호화 할 것 (암호화 시간 단축)
• Header : 암호화 파일 시그니처 (복호화 후에 검증) + 날짜/시간 데이터
• Body : RC4 키를 개인키로 암호화 + 본문 내용을 RC4 키로 암호화
• Tailer : 본문을 md5로 n번 해시 후에 RC4 개인키로 암호화 (무결성 인증을 위해 둠)
• 헤더에 날짜와 시간 값을 저장한 이유는 앞으로 생성된 수 많은 플러그인 엔진 중 하나만 교체되어도
• 백신은 그 플러그인 엔진의 생성 날짜와 시간을 통해 최종적으로 언제 빌드(업데이트) 되었는지 표시 가능
• 암호화 한 내용을 pyc로 컴파일 한 후에 KMD 파일로 저장 (소스 해석 불가)

 

플러그인 엔진 동적 로딩

1. 필요한 플러그인 복호화.
2. 복호화 한 플러그인은 파일로 저장하지 않고 변수에 할당해서 메모리에 올려둠.
3. 현재 pyc 코드이기 때문에 실행 가능한 상태가 아님.
4. imp.new_module(모듈명) 을 통해 빈 모듈을 생성.
5. 따라서 exec(pyc code, 모듈) 을 통해 파이썬 코드와 모듈을 연결
6. sys[모듈명] = 모듈로 전역에서 사용가능하도록 지정하여 사용

 

커널

• 플러그인 로딩 순서
  • 플러그인 엔진의 동작 시간을 고려 (플러그인의 로딩 순서에 따라 전체 시간에 영향을 주는 경우)
  • 플러그인 로딩 순서를 (직접) 기록한 리스트 파일을 참조하여 메모리 로딩 후 작업 수행
  • 리스트 파일 또한 암호화된 상태로 보관
• 플러그인 로딩
  • 플러그인은 pyc 파일이 암호화되어져서 보관.
  • 복호화 후에 marshal 모듈로 로드
  • imp로 새 모듈 생성하고 생성된 모듈명.__dict__와 pyc 코드를 exec를 통해 연결
  • 새 모듈을 플러그인 모듈 처럼 사용가능
• 플러그인 모듈 로딩과 로딩된 플러그인으로부터 인스턴스를 생성하는 것은 별개로 동작
  • 로딩과 인스턴스 생성을 같이 수행 시에 시간이 오래걸림
  • 커널에서는 로드한 플러그인 모듈들과, 인스턴스 들을 각각 객체 리스트에 저장
• 커널 로직
  • 커널 모듈에서 플러그인 모듈 로딩 메소드는 전역으로 둠 (단일 기능)
  • 로딩된 플러그인 모듈은 전역 리스트로 저장
  • 인스턴스 생성 및 작업은 EngineInstance와 같은 명으로 클래스 내부 로직을 통해 수행
  • EngineInstance의 init에서 플러그인 모듈을 통해 각 인스턴스들을 불러오고, 각 인스턴스를 init 시킴
    • 초기화 성공 인스턴스 개수 카운팅 후 출력
    • 초기화 실패 시 인스턴스 목록에 추가하지 않고 pass
      • 따라서 init이 하는일이 없는 모듈의 경우에도 제거하면 안됨
  • EngineInstance의 uninit은 각 인스턴스 uninit 시킴
  • 이후 info 획득도 같은 방식으로 for문을 돌면서 획득
    • init된 인스턴스나 추가한 info 정보도 객체 변수로 저장
      • 이후 다시 메소드 호출하지 않고 객체 변수로 참조하기 위함
  • 치료/ 검사 가능한 virus list를 불러오는 경우
    • 모든 내용을 객체 변수로 저장하면 바이러스가 추가되는 경우 부담이 너무 커짐
    • 따라서 callback 메소드를 파라미터로 넘겨서 list를 받아온 후 해당 callback 메소드로 처리하도록 설계
      • 예를 들어 list 내용을 출력하는 print_virus_list 메소드를 외부에 두고 EngineInstance의 get_list_virus(self, *callback)을 호출 할 때, get_list_virus(print_virus_list)로 호출.
      • get_list_virus에서는 callback 함수가 있다면 객체 변수에 저장하지 않고 callback 함수에 넘겨서 처리하고 callback 함수가 없다면 객체 변수에 저장해서 반환
  • 악성코드 스캔
    • 악성코드 의심 파일을 열때는 fp filehandle이 아닌 mmap을 사용
      • 모든 플러그인에서 파일을 열고 닫으면 시간 소요, handle로 처리하는게 빠름
      • fp 사용하면 이전 plugin에서 read 후에 다음 plugin에서 offset 초기화해야함
    • 악성코드 검사 순서는 플러그인 로딩 순서 기록한 리스트 순서대로 인스턴스가 생성되어 객체 변수에 저장되어 있으므로 해당 순서로 진행
      • 만약 악성코드가 특정 인스턴스에서 검출되면 break하고 인덱스 번호 return
  • 악성코드 치료
    • filename과 scan 후 리턴 받은 uid, 그리고 break된 시점의 engine 인덱스를 통해 치료 요청

 

공유 라이브러리 모듈

• 플러그인 모듈 소스를 빈번하게 수정하는 일을 방지
• 공유 라이브러리 모듈을 각 plugin에서 import 하여 사용
  • 공유 라이브러리 모듈로 사용할 소스도 plugin과 똑같은 포맷의 class를 삽입하여 plugin 형태로 만든다.
  • 공유 라이브러리를 통해 제공할 메소드를 class 외부에 선언한다.
    • 플러그인 로딩 순서를 기록한 문서의 위쪽에 기록하여 커널에 의해 로딩되도록 수행
    • 공유 라이브러리 모듈이 복호화 되고 메모리 load되면서 load되는 모듈은 해당 모듈명으로 sys.module에 의해 전역에서 사용가능하도록 저장되므로 다른 plugin에서 사용 가능.
    • sys.module은 각 모듈에서 import 모듈할 때 가장 먼저 찾는 위치 

 

 

플러그인 엔진 모듈

ex) eicar.py (eicar 악성코드 담당 플러그인 모듈)

class KavMain:
    # eicar는 hash를 통해 검출
    def init(self, plugints_path):
        self.virus_name = 'Eicar-Test-File (not a virus)'
        self.eicar_hash = '44das87asd8zxc9asd89zv6z7xv'
        return 0
    
    def uninit(self):
        del self.virus_name
        del self.dummy_pattern
        return 0

    def scan(self, filehandle, filename):
        try:
            size = os.path.getsize(filename)
            # eicar size와 일치하는 파일 대상으로만 해시를 구함
            if size == 68:
                m = hashlib.md5()
                '''
                fp는 read한 만큼 offset이 이동
                따라서 다음 플러그인에서 fp.seek(0)으로 매번 초기화 해야함
                해결책 filehandle은 fp대신 mmap 사용
                '''
                '''
                import mmap
                fp = open('dummy.txt','rb')
                mm = mmap.mmap(fp.fileno(), 0, access=mmap.ACCESS_READ)
                
                mm[시작위치:시작위치 + 읽을크기]
                mm[:68] 사직위치 생략 시 0으로 취급
                '''
                m.update(filehandle[:68])
                fmd5 = m.hexidigest()
                if fmd5 == self.eicar_hash:
                    return True, self.virus_name, 0
        except IOError:
            pass
        return False, '', -1
    
    def disinfect(self, filename, malware_id):
        try:
            if malware_id == 0 :
                os.remove(filename)
                return True
            # id 값에 따라 치료법 분리 가능
        except IOError:
            pass
        return False    
        
    def list_virus(self):
        vlist = list()
        vlist.append(self.virus_name)
        return vlist
    
    def get_info(self):
        info = dict()
        info['author'] = "HJ LIM"
        info['version'] = '1.0'
        info['title'] = 'Eicar Test Engine'

https://github.com/hanul93/kicomav/