chg: Use typing info of lief

pull/1144/head
Raphaël Vinot 2024-01-31 12:15:08 +01:00
parent ec11c1ee0c
commit ca0fb8dc99
4 changed files with 146 additions and 80 deletions

View File

@ -2,12 +2,13 @@
from __future__ import annotations
from io import BytesIO
from . import FileObject
from ..exceptions import MISPObjectException
import logging
from io import BytesIO
from typing import Any
from . import FileObject, PEObject, ELFObject, MachOObject, PESectionObject, ELFSectionObject, MachOSectionObject
from ..exceptions import MISPObjectException
logger = logging.getLogger('pymisp')
try:
@ -32,7 +33,11 @@ class FileTypeNotImplemented(MISPObjectException):
pass
def make_binary_objects(filepath: str | None = None, pseudofile: BytesIO | bytes | None = None, filename: str | None = None, standalone: bool = True, default_attributes_parameters: dict = {}):
def make_binary_objects(filepath: str | None = None,
pseudofile: BytesIO | bytes | None = None,
filename: str | None = None,
standalone: bool = True,
default_attributes_parameters: dict[str, Any] = {}) -> tuple[FileObject, PEObject | ELFObject | MachOObject | None, list[PESectionObject] | list[ELFSectionObject] | list[MachOSectionObject]]:
misp_file = FileObject(filepath=filepath, pseudofile=pseudofile, filename=filename,
standalone=standalone, default_attributes_parameters=default_attributes_parameters)
if HAS_LIEF and (filepath or pseudofile):

View File

@ -2,13 +2,16 @@
from __future__ import annotations
import logging
from hashlib import md5, sha1, sha256, sha512
from io import BytesIO
from pathlib import Path
from typing import Any
from . import FileObject
from .abstractgenerator import AbstractMISPObjectGenerator
from ..exceptions import InvalidMISPObject
from io import BytesIO
from hashlib import md5, sha1, sha256, sha512
import logging
from pathlib import Path
from . import FileObject
import lief
@ -21,7 +24,10 @@ except ImportError:
logger = logging.getLogger('pymisp')
def make_elf_objects(lief_parsed: lief.ELF.Binary, misp_file: FileObject, standalone: bool = True, default_attributes_parameters: dict = {}):
def make_elf_objects(lief_parsed: lief.ELF.Binary,
misp_file: FileObject,
standalone: bool = True,
default_attributes_parameters: dict[str, Any] = {}) -> tuple[FileObject, ELFObject, list[ELFSectionObject]]:
elf_object = ELFObject(parsed=lief_parsed, standalone=standalone, default_attributes_parameters=default_attributes_parameters)
misp_file.add_reference(elf_object.uuid, 'includes', 'ELF indicators')
elf_sections = []
@ -32,22 +38,30 @@ def make_elf_objects(lief_parsed: lief.ELF.Binary, misp_file: FileObject, standa
class ELFObject(AbstractMISPObjectGenerator):
def __init__(self, parsed: lief.ELF.Binary | None = None, filepath: Path | str | None = None, pseudofile: BytesIO | bytes | list[int] | None = None, **kwargs):
__elf: lief.ELF.Binary
def __init__(self, parsed: lief.ELF.Binary | None = None, # type: ignore[no-untyped-def]
filepath: Path | str | None = None,
pseudofile: BytesIO | bytes | list[int] | None = None, **kwargs) -> None:
"""Creates an ELF object, with lief"""
super().__init__('elf', **kwargs)
if not HAS_PYDEEP:
logger.warning("pydeep is missing, please install pymisp this way: pip install pymisp[fileobjects]")
if pseudofile:
if isinstance(pseudofile, BytesIO):
self.__elf = lief.ELF.parse(obj=pseudofile)
e = lief.ELF.parse(obj=pseudofile)
elif isinstance(pseudofile, bytes):
self.__elf = lief.ELF.parse(raw=list(pseudofile))
e = lief.ELF.parse(raw=list(pseudofile))
elif isinstance(pseudofile, list):
self.__elf = lief.ELF.parse(raw=pseudofile)
e = lief.ELF.parse(raw=pseudofile)
else:
raise InvalidMISPObject(f'Pseudo file can be BytesIO or bytes got {type(pseudofile)}')
if not e:
raise InvalidMISPObject('Unable to parse pseudofile')
self.__elf = e
elif filepath:
self.__elf = lief.ELF.parse(filepath)
if e := lief.ELF.parse(filepath):
self.__elf = e
elif parsed:
# Got an already parsed blob
if isinstance(parsed, lief.ELF.Binary):
@ -56,7 +70,7 @@ class ELFObject(AbstractMISPObjectGenerator):
raise InvalidMISPObject(f'Not a lief.ELF.Binary: {type(parsed)}')
self.generate_attributes()
def generate_attributes(self):
def generate_attributes(self) -> None:
# General information
self.add_attribute('type', value=str(self.__elf.header.file_type).split('.')[1])
self.add_attribute('entrypoint-address', value=self.__elf.entrypoint)
@ -78,7 +92,7 @@ class ELFObject(AbstractMISPObjectGenerator):
class ELFSectionObject(AbstractMISPObjectGenerator):
def __init__(self, section: lief.ELF.Section, **kwargs):
def __init__(self, section: lief.ELF.Section, **kwargs) -> None: # type: ignore[no-untyped-def]
"""Creates an ELF Section object. Object generated by ELFObject."""
# Python3 way
# super().__init__('pe-section')
@ -87,13 +101,13 @@ class ELFSectionObject(AbstractMISPObjectGenerator):
self.__data = bytes(self.__section.content)
self.generate_attributes()
def generate_attributes(self):
def generate_attributes(self) -> None:
self.add_attribute('name', value=self.__section.name)
self.add_attribute('type', value=str(self.__section.type).split('.')[1])
for flag in self.__section.flags_list:
self.add_attribute('flag', value=str(flag).split('.')[1])
size = self.add_attribute('size-in-bytes', value=self.__section.size)
if int(size.value) > 0:
self.add_attribute('size-in-bytes', value=self.__section.size)
if int(self.__section.size) > 0:
self.add_attribute('entropy', value=self.__section.entropy)
self.add_attribute('md5', value=md5(self.__data).hexdigest())
self.add_attribute('sha1', value=sha1(self.__data).hexdigest())

View File

@ -2,13 +2,17 @@
from __future__ import annotations
from ..exceptions import InvalidMISPObject
from .abstractgenerator import AbstractMISPObjectGenerator
from io import BytesIO
from hashlib import md5, sha1, sha256, sha512
import logging
from hashlib import md5, sha1, sha256, sha512
from io import BytesIO
from pathlib import Path
from typing import Any
from ..exceptions import InvalidMISPObject
from . import FileObject
from .abstractgenerator import AbstractMISPObjectGenerator
import lief
@ -21,7 +25,10 @@ except ImportError:
logger = logging.getLogger('pymisp')
def make_macho_objects(lief_parsed: lief.MachO.Binary, misp_file: FileObject, standalone: bool = True, default_attributes_parameters: dict = {}):
def make_macho_objects(lief_parsed: lief.MachO.Binary,
misp_file: FileObject,
standalone: bool = True,
default_attributes_parameters: dict[str, Any] = {}) -> tuple[FileObject, MachOObject, list[MachOSectionObject]]:
macho_object = MachOObject(parsed=lief_parsed, standalone=standalone, default_attributes_parameters=default_attributes_parameters)
misp_file.add_reference(macho_object.uuid, 'includes', 'MachO indicators')
macho_sections = []
@ -32,33 +39,43 @@ def make_macho_objects(lief_parsed: lief.MachO.Binary, misp_file: FileObject, st
class MachOObject(AbstractMISPObjectGenerator):
def __init__(self, parsed: lief.MachO.Binary | None = None, filepath: Path | str | None = None, pseudofile: BytesIO | list[int] | None = None, **kwargs):
__macho: lief.MachO.Binary
def __init__(self, parsed: lief.MachO.Binary | lief.MachO.FatBinary | None = None, # type: ignore[no-untyped-def]
filepath: Path | str | None = None,
pseudofile: BytesIO | list[int] | None = None,
**kwargs) -> None:
"""Creates an MachO object, with lief"""
super().__init__('macho', **kwargs)
if not HAS_PYDEEP:
logger.warning("pydeep is missing, please install pymisp this way: pip install pymisp[fileobjects]")
if pseudofile:
if isinstance(pseudofile, BytesIO):
self.__macho = lief.MachO.parse(obj=pseudofile)
m = lief.MachO.parse(obj=pseudofile)
elif isinstance(pseudofile, bytes):
self.__macho = lief.MachO.parse(raw=list(pseudofile))
m = lief.MachO.parse(raw=list(pseudofile))
elif isinstance(pseudofile, list):
self.__macho = lief.MachO.parse(raw=pseudofile)
m = lief.MachO.parse(raw=pseudofile)
else:
raise InvalidMISPObject(f'Pseudo file can be BytesIO or bytes got {type(pseudofile)}')
if not m:
raise InvalidMISPObject('Unable to parse pseudofile')
self.__macho = m.at(0)
elif filepath:
self.__macho = lief.MachO.parse(filepath)
if m := lief.MachO.parse(filepath):
self.__macho = m.at(0)
elif parsed:
# Got an already parsed blob
if isinstance(parsed, lief.MachO.FatBinary):
self.__macho = parsed.at(0)
elif isinstance(parsed, lief.MachO.Binary):
self.__macho = parsed
else:
raise InvalidMISPObject(f'Not a lief.MachO.Binary: {type(parsed)}')
self.generate_attributes()
def generate_attributes(self):
def generate_attributes(self) -> None:
self.add_attribute('type', value=str(self.__macho.header.file_type).split('.')[1])
self.add_attribute('name', value=self.__macho.name)
# General information
if self.__macho.has_entrypoint:
self.add_attribute('entrypoint-address', value=self.__macho.entrypoint)
@ -76,7 +93,7 @@ class MachOObject(AbstractMISPObjectGenerator):
class MachOSectionObject(AbstractMISPObjectGenerator):
def __init__(self, section: lief.MachO.Section, **kwargs):
def __init__(self, section: lief.MachO.Section, **kwargs) -> None: # type: ignore[no-untyped-def]
"""Creates an MachO Section object. Object generated by MachOObject."""
# Python3 way
# super().__init__('pe-section')
@ -85,10 +102,10 @@ class MachOSectionObject(AbstractMISPObjectGenerator):
self.__data = bytes(self.__section.content)
self.generate_attributes()
def generate_attributes(self):
def generate_attributes(self) -> None:
self.add_attribute('name', value=self.__section.name)
size = self.add_attribute('size-in-bytes', value=self.__section.size)
if int(size.value) > 0:
self.add_attribute('size-in-bytes', value=self.__section.size)
if int(self.__section.size) > 0:
self.add_attribute('entropy', value=self.__section.entropy)
self.add_attribute('md5', value=md5(self.__data).hexdigest())
self.add_attribute('sha1', value=sha1(self.__data).hexdigest())

View File

@ -2,16 +2,18 @@
from __future__ import annotations
from ..exceptions import InvalidMISPObject
from .abstractgenerator import AbstractMISPObjectGenerator
from io import BytesIO
from hashlib import md5, sha1, sha256, sha512
from datetime import datetime
import logging
from pathlib import Path
from base64 import b64encode
from datetime import datetime
from hashlib import md5, sha1, sha256, sha512
from io import BytesIO
from pathlib import Path
from typing import Any
from . import FileObject
from .abstractgenerator import AbstractMISPObjectGenerator
from ..exceptions import InvalidMISPObject
import lief
@ -24,7 +26,10 @@ except ImportError:
logger = logging.getLogger('pymisp')
def make_pe_objects(lief_parsed: lief.PE.Binary, misp_file: FileObject, standalone: bool = True, default_attributes_parameters: dict = {}):
def make_pe_objects(lief_parsed: lief.PE.Binary,
misp_file: FileObject,
standalone: bool = True,
default_attributes_parameters: dict[str, Any] = {}) -> tuple[FileObject, PEObject, list[PESectionObject]]:
pe_object = PEObject(parsed=lief_parsed, standalone=standalone, default_attributes_parameters=default_attributes_parameters)
misp_file.add_reference(pe_object.uuid, 'includes', 'PE indicators')
pe_sections = []
@ -35,22 +40,33 @@ def make_pe_objects(lief_parsed: lief.PE.Binary, misp_file: FileObject, standalo
class PEObject(AbstractMISPObjectGenerator):
def __init__(self, parsed: lief.PE.Binary | None = None, filepath: Path | str | None = None, pseudofile: BytesIO | list[int] | None = None, **kwargs):
__pe: lief.PE.Binary
def __init__(self, parsed: lief.PE.Binary | None = None, # type: ignore[no-untyped-def]
filepath: Path | str | None = None,
pseudofile: BytesIO | list[int] | None = None,
**kwargs) -> None:
"""Creates an PE object, with lief"""
super().__init__('pe', **kwargs)
if not HAS_PYDEEP:
logger.warning("pydeep is missing, please install pymisp this way: pip install pymisp[fileobjects]")
if pseudofile:
if isinstance(pseudofile, BytesIO):
self.__pe = lief.PE.parse(obj=pseudofile)
p = lief.PE.parse(obj=pseudofile)
elif isinstance(pseudofile, bytes):
self.__pe = lief.PE.parse(raw=list(pseudofile))
p = lief.PE.parse(raw=list(pseudofile))
elif isinstance(pseudofile, list):
self.__pe = lief.PE.parse(raw=pseudofile)
p = lief.PE.parse(raw=pseudofile)
else:
raise InvalidMISPObject(f'Pseudo file can be BytesIO or bytes got {type(pseudofile)}')
if not p:
raise InvalidMISPObject('Unable to parse pseudofile')
self.__pe = p
elif filepath:
self.__pe = lief.PE.parse(filepath)
if p := lief.PE.parse(filepath):
self.__pe = p
else:
raise InvalidMISPObject(f'Unable to parse {filepath}')
elif parsed:
# Got an already parsed blob
if isinstance(parsed, lief.PE.Binary):
@ -59,22 +75,22 @@ class PEObject(AbstractMISPObjectGenerator):
raise InvalidMISPObject(f'Not a lief.PE.Binary: {type(parsed)}')
self.generate_attributes()
def _is_exe(self):
def _is_exe(self) -> bool:
if not self._is_dll() and not self._is_driver():
return self.__pe.header.has_characteristic(lief.PE.HEADER_CHARACTERISTICS.EXECUTABLE_IMAGE)
return self.__pe.header.has_characteristic(lief.PE.Header.CHARACTERISTICS.EXECUTABLE_IMAGE)
return False
def _is_dll(self):
return self.__pe.header.has_characteristic(lief.PE.HEADER_CHARACTERISTICS.DLL)
def _is_dll(self) -> bool:
return self.__pe.header.has_characteristic(lief.PE.Header.CHARACTERISTICS.DLL)
def _is_driver(self):
def _is_driver(self) -> bool:
# List from pefile
system_DLLs = {'ntoskrnl.exe', 'hal.dll', 'ndis.sys', 'bootvid.dll', 'kdcom.dll'}
if system_DLLs.intersection([imp.lower() for imp in self.__pe.libraries]):
return True
return False
def _get_pe_type(self):
def _get_pe_type(self) -> str:
if self._is_dll():
return 'dll'
elif self._is_driver():
@ -84,31 +100,27 @@ class PEObject(AbstractMISPObjectGenerator):
else:
return 'unknown'
def generate_attributes(self):
def generate_attributes(self) -> None:
self.add_attribute('type', value=self._get_pe_type())
# General information
self.add_attribute('entrypoint-address', value=self.__pe.entrypoint)
self.add_attribute('compilation-timestamp', value=datetime.utcfromtimestamp(self.__pe.header.time_date_stamps).isoformat())
self.add_attribute('imphash', value=lief.PE.get_imphash(self.__pe, lief.PE.IMPHASH_MODE.PEFILE))
self.add_attribute('authentihash', value=self.__pe.authentihash_sha256.hex())
try:
if (self.__pe.has_resources
and self.__pe.resources_manager.has_version
and self.__pe.resources_manager.version.has_string_file_info
and self.__pe.resources_manager.version.string_file_info.langcode_items):
fileinfo = dict(self.__pe.resources_manager.version.string_file_info.langcode_items[0].items.items())
r_manager = self.__pe.resources_manager
if isinstance(r_manager, lief.PE.ResourcesManager):
version = r_manager.version
if isinstance(version, lief.PE.ResourceVersion) and version.string_file_info is not None:
fileinfo = dict(version.string_file_info.langcode_items[0].items.items())
self.add_attribute('original-filename', value=fileinfo.get('OriginalFilename'))
self.add_attribute('internal-filename', value=fileinfo.get('InternalName'))
self.add_attribute('file-description', value=fileinfo.get('FileDescription'))
self.add_attribute('file-version', value=fileinfo.get('FileVersion'))
self.add_attribute('lang-id', value=self.__pe.resources_manager.version.string_file_info.langcode_items[0].key)
self.add_attribute('product-name', value=fileinfo.get('ProductName'))
self.add_attribute('product-version', value=fileinfo.get('ProductVersion'))
self.add_attribute('company-name', value=fileinfo.get('CompanyName'))
self.add_attribute('legal-copyright', value=fileinfo.get('LegalCopyright'))
except lief.read_out_of_bound:
# The file is corrupted
pass
self.add_attribute('lang-id', value=version.string_file_info.langcode_items[0].key)
# Sections
self.sections = []
if self.__pe.sections:
@ -121,7 +133,11 @@ class PEObject(AbstractMISPObjectGenerator):
self.add_reference(s.uuid, 'includes', f'Section {pos} of PE')
if ((self.__pe.entrypoint >= section.virtual_address)
and (self.__pe.entrypoint < (section.virtual_address + section.virtual_size))):
self.add_attribute('entrypoint-section-at-position', value=f'{section.name}|{pos}')
if isinstance(section.name, bytes):
section_name = section.name.decode()
else:
section_name = section.name
self.add_attribute('entrypoint-section-at-position', value=f'{section_name}|{pos}')
pos += 1
self.sections.append(s)
self.add_attribute('number-sections', value=len(self.sections))
@ -141,16 +157,30 @@ class PEObject(AbstractMISPObjectGenerator):
class PECertificate(AbstractMISPObjectGenerator):
def __init__(self, certificate: lief.PE.x509, **kwargs):
def __init__(self, certificate: lief.PE.x509, **kwargs) -> None: # type: ignore[no-untyped-def]
super().__init__('x509')
self.__certificate = certificate
self.generate_attributes()
def generate_attributes(self):
def generate_attributes(self) -> None:
self.add_attribute('issuer', value=self.__certificate.issuer)
self.add_attribute('serial-number', value=self.__certificate.serial_number)
self.add_attribute('validity-not-before', value=datetime(*self.__certificate.valid_from))
self.add_attribute('validity-not-after', value=datetime(*self.__certificate.valid_to))
if len(self.__certificate.valid_from) == 6:
self.add_attribute('validity-not-before',
value=datetime(year=self.__certificate.valid_from[0],
month=self.__certificate.valid_from[1],
day=self.__certificate.valid_from[2],
hour=self.__certificate.valid_from[3],
minute=self.__certificate.valid_from[4],
second=self.__certificate.valid_from[5]))
if len(self.__certificate.valid_to) == 6:
self.add_attribute('validity-not-after',
value=datetime(year=self.__certificate.valid_to[0],
month=self.__certificate.valid_to[1],
day=self.__certificate.valid_to[2],
hour=self.__certificate.valid_to[3],
minute=self.__certificate.valid_to[4],
second=self.__certificate.valid_to[5]))
self.add_attribute('version', value=self.__certificate.version)
self.add_attribute('subject', value=self.__certificate.subject)
self.add_attribute('signature_algorithm', value=self.__certificate.signature_algorithm)
@ -159,19 +189,19 @@ class PECertificate(AbstractMISPObjectGenerator):
class PESigners(AbstractMISPObjectGenerator):
def __init__(self, signer: lief.PE.SignerInfo, **kwargs):
def __init__(self, signer: lief.PE.SignerInfo, **kwargs) -> None: # type: ignore[no-untyped-def]
super().__init__('authenticode-signerinfo')
self.__signer = signer
self.generate_attributes()
def generate_attributes(self):
def generate_attributes(self) -> None:
self.add_attribute('issuer', value=self.__signer.issuer)
self.add_attribute('serial-number', value=self.__signer.serial_number)
self.add_attribute('version', value=self.__signer.version)
self.add_attribute('digest_algorithm', value=self.__signer.digest_algorithm.name)
self.add_attribute('encryption_algorithm', value=self.__signer.encryption_algorithm.name)
self.add_attribute('digest_algorithm', value=str(self.__signer.digest_algorithm))
self.add_attribute('encryption_algorithm', value=str(self.__signer.encryption_algorithm))
self.add_attribute('digest-base64', value=b64encode(self.__signer.encrypted_digest))
info = self.__signer.get_attribute(lief.PE.SIG_ATTRIBUTE_TYPES.SPC_SP_OPUS_INFO)
info: lief.PE.SpcSpOpusInfo = self.__signer.get_attribute(lief.PE.SIG_ATTRIBUTE_TYPES.SPC_SP_OPUS_INFO) # type: ignore[attr-defined, assignment]
if info:
self.add_attribute('program-name', value=info.program_name)
self.add_attribute('url', value=info.more_info)
@ -179,17 +209,17 @@ class PESigners(AbstractMISPObjectGenerator):
class PESectionObject(AbstractMISPObjectGenerator):
def __init__(self, section: lief.PE.Section, **kwargs):
def __init__(self, section: lief.PE.Section, **kwargs) -> None: # type: ignore[no-untyped-def]
"""Creates an PE Section object. Object generated by PEObject."""
super().__init__('pe-section')
self.__section = section
self.__data = bytes(self.__section.content)
self.generate_attributes()
def generate_attributes(self):
def generate_attributes(self) -> None:
self.add_attribute('name', value=self.__section.name)
size = self.add_attribute('size-in-bytes', value=self.__section.size)
if int(size.value) > 0:
self.add_attribute('size-in-bytes', value=self.__section.size)
if int(self.__section.size) > 0:
# zero-filled sections can create too many correlations
to_ids = float(self.__section.entropy) > 0
disable_correlation = not to_ids