<commit_before>#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import sys
import json
import os
import datetime
import re
import ntpath
import socket
from misp2stix_mapping import *
from collections import defaultdict
from copy import deepcopy
from cybox.core import Object, Observable, ObservableComposition, RelatedObject
from cybox.common import Hash, HashList, ByteRun, ByteRuns
from cybox.common.object_properties import CustomProperties,  Property
from cybox.objects.account_object import Account, Authentication, StructuredAuthenticationMechanism
from cybox.objects.address_object import Address
from cybox.objects.artifact_object import Artifact, RawArtifact
from cybox.objects.as_object import AutonomousSystem
from cybox.objects.custom_object import Custom
from cybox.objects.domain_name_object import DomainName
from cybox.objects.email_message_object import EmailMessage, EmailHeader, EmailRecipients, Attachments
from cybox.objects.file_object import File
from cybox.objects.hostname_object import Hostname
from cybox.objects.http_session_object import HTTPClientRequest, HTTPRequestHeader, HTTPRequestHeaderFields, HTTPRequestLine, HTTPRequestResponse, HTTPSession
from cybox.objects.mutex_object import Mutex
from cybox.objects.network_connection_object import NetworkConnection
from cybox.objects.network_socket_object import NetworkSocket
from cybox.objects.pipe_object import Pipe
from cybox.objects.port_object import Port
from cybox.objects.process_object import Process
from cybox.objects.socket_address_object import SocketAddress
from cybox.objects.system_object import System, NetworkInterface, NetworkInterfaceList
from cybox.objects.unix_user_account_object import UnixUserAccount
from cybox.objects.uri_object import URI
from cybox.objects.user_account_object import UserAccount
from cybox.objects.whois_object import WhoisEntry, WhoisRegistrants, WhoisRegistrant, WhoisRegistrar, WhoisNameservers
from cybox.objects.win_executable_file_object import WinExecutableFile, PEHeaders, PEFileHeader, PESectionList, PESection, PESectionHeaderStruct, Entropy
from cybox.objects.win_registry_key_object import RegistryValue, RegistryValues, WinRegistryKey
from cybox.objects.win_service_object import WinService
from cybox.objects.win_user_object import WinUser
from cybox.objects.x509_certificate_object import X509Certificate, X509CertificateSignature, X509Cert, SubjectPublicKey, RSAPublicKey, Validity
from cybox.utils import Namespace
from stix.coa import CourseOfAction
from stix.common import InformationSource, Identity, ToolInformation
from stix.common.confidence import Confidence
from stix.common.related import RelatedIndicator, RelatedObservable, RelatedThreatActor, RelatedTTP
from stix.common.vocabs import IncidentStatus
from stix.core import STIXPackage, STIXHeader
from stix.data_marking import Marking, MarkingSpecification
from stix.exploit_target import ExploitTarget, Vulnerability, Weakness
from stix.exploit_target.vulnerability import CVSSVector
from stix.extensions.identity.ciq_identity_3_0 import CIQIdentity3_0Instance, STIXCIQIdentity3_0, PartyName, ElectronicAddressIdentifier, FreeTextAddress
from stix.extensions.identity.ciq_identity_3_0 import Address as ciq_Address
from stix.extensions.marking.tlp import TLPMarkingStructure
from stix.extensions.test_mechanism.snort_test_mechanism import SnortTestMechanism
from stix.incident import Incident, Time, ExternalID, AffectedAsset, AttributedThreatActors
from stix.incident.history import History, HistoryItem
from stix.indicator import Indicator
from stix.indicator.valid_time import ValidTime
from stix.threat_actor import ThreatActor
from stix.ttp import TTP, Behavior
from stix.ttp.attack_pattern import AttackPattern
from stix.ttp.malware_instance import MalwareInstance
from stix.ttp.resource import Resource, Tools

try:
    from stix.utils import idgen
except ImportError:
    from mixbox import idgen

namespace = ['https://github.com/MISP/MISP', 'MISP']

this_module = sys.modules[__name__]


class StixBuilder(object):
    def __init__(self, args):
        self.args = args
        self.return_type = args[2]
        self.baseurl = args[3] if (len(args) > 3 and args[3]) else namespace[0]
        self.orgname = re.sub('[\W]+', '', args[4].replace(" ", "_")) if (len(args) > 4 and args[4]) else namespace[1]
        try:
            idgen.set_id_namespace({self.baseurl: self.orgname})
        except ValueError:
            try:
                idgen.set_id_namespace(Namespace(self.baseurl, self.orgname))
            except TypeError:
                idgen.set_id_namespace(Namespace(self.baseurl, self.orgname, "MISP"))
        self.namespace_prefix = idgen.get_id_namespace_alias()
        ## MAPPING FOR ATTRIBUTES
        self.simple_type_to_method = {"port": self.generate_port_observable, "domain|ip": self.generate_domain_ip_observable,
                                      "named pipe": self.generate_pipe_observable}
        self.simple_type_to_method.update(dict.fromkeys(list(hash_type_attributes["single"]) + list(hash_type_attributes["composite"]) + ["filename"], self.resolve_file_observable))
        self.simple_type_to_method.update(dict.fromkeys(["ip-src", "ip-dst"], self.generate_ip_observable))
        self.simple_type_to_method.update(dict.fromkeys(["ip-src|port", "ip-dst|port", "hostname|port"], self.generate_socket_address_observable))
        self.simple_type_to_method.update(dict.fromkeys(["regkey", "regkey|value"], self.generate_regkey_observable))
        self.simple_type_to_method.update(dict.fromkeys(["hostname", "domain", "url", "AS", "mutex", "named pipe", "link", "windows-service-name"], self.generate_simple_observable))
        self.simple_type_to_method.update(dict.fromkeys(["email-src", "email-dst", "email-subject", "email-reply-to"], self.resolve_email_observable))
        self.simple_type_to_method.update(dict.fromkeys(["http-method", "user-agent"], self.resolve_http_observable))
        self.simple_type_to_method.update(dict.fromkeys(["pattern-in-file", "pattern-in-traffic", "pattern-in-memory"], self.resolve_pattern_observable))
        self.simple_type_to_method.update(dict.fromkeys(["mac-address"], self.resolve_system_observable))
        self.simple_type_to_method.update(dict.fromkeys(["attachment"], self.resolve_attachment))
        self.simple_type_to_method.update(dict.fromkeys(["email-attachment"], self.generate_email_attachment_observable))
        self.simple_type_to_method.update(dict.fromkeys(["malware-sample"], self.resolve_malware_sample))
        ## MAPPING FOR OBJECTS
        self.ttp_names = {'attack-pattern': self.parse_attack_pattern,
                          'course-of-action': self.parse_course_of_action,
                          'vulnerability': self.parse_vulnerability,
                          'weakness': self.parse_weakness}
        self.types_mapping = {CourseOfAction: 'add_course_of_action',
                              ThreatActor: 'add_threat_actor',
                              TTP: 'add_ttp'}
        self.objects_mapping = {"asn": self.parse_asn_object,
                                "credential": self.parse_credential_object,
                                "domain-ip": self.parse_domain_ip_object,
                                "email": self.parse_email_object,
                                "file": self.parse_file_object,
                                "ip-port": self.parse_ip_port_object,
                                "network-connection": self.parse_network_connection_object,
                                "network-socket": self.parse_network_socket_object,
                                "pe": self.store_pe,
                                "pe-section": self.store_pe,
                                "process": self.parse_process_object,
                                "registry-key": self.parse_regkey_object,
                                "url": self.parse_url_object,
                                "user-account": self.parse_user_account_object,
                                "whois": self.parse_whois,
                                "x509": self.parse_x509_object}

    def loadEvent(self):
        pathname = os.path.dirname(self.args[0])
        self.filename = "{}/tmp/{}".format(pathname, self.args[1])
        with open(self.filename, 'rt', encoding='utf-8') as f:
            self.json_event = json.loads(f.read())

    def generateEventPackages(self):
        try
            return_type_to_package = {'json': ('to_json', {}),
                                      'xml': ('to_xml', {'include_namespaces': False, 'include_schemalocs': False, 'encoding': 'utf8'})}
            to_call, args = return_type_to_package[self.return_type]
            separator = None
            if self.json_event.get('response'):
                from misp_framing import stix_framing
                _, separator, _ = stix_framing(self.baseurl, self.orgname, self.return_type)
                stix_packages = [getattr(self.generate_package(event['Event']), to_call)(**args) for event in self.json_event['response']]
            else:
                stix_packages = [getattr(self.generate_package(self.json_event['Event']), to_call)(**args)]
            if self.return_type == 'xml':
                stix_packages = [s.decode() for s in stix_packages]
                stix_packages = ['\n            '.join(s.split('\n')[:-1]).replace('stix:STIX_Package', 'stix:Package') for s in stix_packages]
                stix_packages = ['            {}\n'.format(s) for s in stix_packages]
            else:
                stix_packages = ['{"package": %s}' % s for s in stix_packages]
            self.stix_package = separator.join(stix_packages) if len(stix_packages) > 1 else stix_packages[0]
            self.saveFile()
            print(json.dumps({'success': 1}))
        except Exception as e:
            print(json.dumps({'error': e.__str__()}))
<commit_msg>fix: [stix export] Typo<commit_after>#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import sys
import json
import os
import datetime
import re
import ntpath
import socket
from misp2stix_mapping import *
from collections import defaultdict
from copy import deepcopy
from cybox.core import Object, Observable, ObservableComposition, RelatedObject
from cybox.common import Hash, HashList, ByteRun, ByteRuns
from cybox.common.object_properties import CustomProperties,  Property
from cybox.objects.account_object import Account, Authentication, StructuredAuthenticationMechanism
from cybox.objects.address_object import Address
from cybox.objects.artifact_object import Artifact, RawArtifact
from cybox.objects.as_object import AutonomousSystem
from cybox.objects.custom_object import Custom
from cybox.objects.domain_name_object import DomainName
from cybox.objects.email_message_object import EmailMessage, EmailHeader, EmailRecipients, Attachments
from cybox.objects.file_object import File
from cybox.objects.hostname_object import Hostname
from cybox.objects.http_session_object import HTTPClientRequest, HTTPRequestHeader, HTTPRequestHeaderFields, HTTPRequestLine, HTTPRequestResponse, HTTPSession
from cybox.objects.mutex_object import Mutex
from cybox.objects.network_connection_object import NetworkConnection
from cybox.objects.network_socket_object import NetworkSocket
from cybox.objects.pipe_object import Pipe
from cybox.objects.port_object import Port
from cybox.objects.process_object import Process
from cybox.objects.socket_address_object import SocketAddress
from cybox.objects.system_object import System, NetworkInterface, NetworkInterfaceList
from cybox.objects.unix_user_account_object import UnixUserAccount
from cybox.objects.uri_object import URI
from cybox.objects.user_account_object import UserAccount
from cybox.objects.whois_object import WhoisEntry, WhoisRegistrants, WhoisRegistrant, WhoisRegistrar, WhoisNameservers
from cybox.objects.win_executable_file_object import WinExecutableFile, PEHeaders, PEFileHeader, PESectionList, PESection, PESectionHeaderStruct, Entropy
from cybox.objects.win_registry_key_object import RegistryValue, RegistryValues, WinRegistryKey
from cybox.objects.win_service_object import WinService
from cybox.objects.win_user_object import WinUser
from cybox.objects.x509_certificate_object import X509Certificate, X509CertificateSignature, X509Cert, SubjectPublicKey, RSAPublicKey, Validity
from cybox.utils import Namespace
from stix.coa import CourseOfAction
from stix.common import InformationSource, Identity, ToolInformation
from stix.common.confidence import Confidence
from stix.common.related import RelatedIndicator, RelatedObservable, RelatedThreatActor, RelatedTTP
from stix.common.vocabs import IncidentStatus
from stix.core import STIXPackage, STIXHeader
from stix.data_marking import Marking, MarkingSpecification
from stix.exploit_target import ExploitTarget, Vulnerability, Weakness
from stix.exploit_target.vulnerability import CVSSVector
from stix.extensions.identity.ciq_identity_3_0 import CIQIdentity3_0Instance, STIXCIQIdentity3_0, PartyName, ElectronicAddressIdentifier, FreeTextAddress
from stix.extensions.identity.ciq_identity_3_0 import Address as ciq_Address
from stix.extensions.marking.tlp import TLPMarkingStructure
from stix.extensions.test_mechanism.snort_test_mechanism import SnortTestMechanism
from stix.incident import Incident, Time, ExternalID, AffectedAsset, AttributedThreatActors
from stix.incident.history import History, HistoryItem
from stix.indicator import Indicator
from stix.indicator.valid_time import ValidTime
from stix.threat_actor import ThreatActor
from stix.ttp import TTP, Behavior
from stix.ttp.attack_pattern import AttackPattern
from stix.ttp.malware_instance import MalwareInstance
from stix.ttp.resource import Resource, Tools

try:
    from stix.utils import idgen
except ImportError:
    from mixbox import idgen

namespace = ['https://github.com/MISP/MISP', 'MISP']

this_module = sys.modules[__name__]


class StixBuilder(object):
    def __init__(self, args):
        self.args = args
        self.return_type = args[2]
        self.baseurl = args[3] if (len(args) > 3 and args[3]) else namespace[0]
        self.orgname = re.sub('[\W]+', '', args[4].replace(" ", "_")) if (len(args) > 4 and args[4]) else namespace[1]
        try:
            idgen.set_id_namespace({self.baseurl: self.orgname})
        except ValueError:
            try:
                idgen.set_id_namespace(Namespace(self.baseurl, self.orgname))
            except TypeError:
                idgen.set_id_namespace(Namespace(self.baseurl, self.orgname, "MISP"))
        self.namespace_prefix = idgen.get_id_namespace_alias()
        ## MAPPING FOR ATTRIBUTES
        self.simple_type_to_method = {"port": self.generate_port_observable, "domain|ip": self.generate_domain_ip_observable,
                                      "named pipe": self.generate_pipe_observable}
        self.simple_type_to_method.update(dict.fromkeys(list(hash_type_attributes["single"]) + list(hash_type_attributes["composite"]) + ["filename"], self.resolve_file_observable))
        self.simple_type_to_method.update(dict.fromkeys(["ip-src", "ip-dst"], self.generate_ip_observable))
        self.simple_type_to_method.update(dict.fromkeys(["ip-src|port", "ip-dst|port", "hostname|port"], self.generate_socket_address_observable))
        self.simple_type_to_method.update(dict.fromkeys(["regkey", "regkey|value"], self.generate_regkey_observable))
        self.simple_type_to_method.update(dict.fromkeys(["hostname", "domain", "url", "AS", "mutex", "named pipe", "link", "windows-service-name"], self.generate_simple_observable))
        self.simple_type_to_method.update(dict.fromkeys(["email-src", "email-dst", "email-subject", "email-reply-to"], self.resolve_email_observable))
        self.simple_type_to_method.update(dict.fromkeys(["http-method", "user-agent"], self.resolve_http_observable))
        self.simple_type_to_method.update(dict.fromkeys(["pattern-in-file", "pattern-in-traffic", "pattern-in-memory"], self.resolve_pattern_observable))
        self.simple_type_to_method.update(dict.fromkeys(["mac-address"], self.resolve_system_observable))
        self.simple_type_to_method.update(dict.fromkeys(["attachment"], self.resolve_attachment))
        self.simple_type_to_method.update(dict.fromkeys(["email-attachment"], self.generate_email_attachment_observable))
        self.simple_type_to_method.update(dict.fromkeys(["malware-sample"], self.resolve_malware_sample))
        ## MAPPING FOR OBJECTS
        self.ttp_names = {'attack-pattern': self.parse_attack_pattern,
                          'course-of-action': self.parse_course_of_action,
                          'vulnerability': self.parse_vulnerability,
                          'weakness': self.parse_weakness}
        self.types_mapping = {CourseOfAction: 'add_course_of_action',
                              ThreatActor: 'add_threat_actor',
                              TTP: 'add_ttp'}
        self.objects_mapping = {"asn": self.parse_asn_object,
                                "credential": self.parse_credential_object,
                                "domain-ip": self.parse_domain_ip_object,
                                "email": self.parse_email_object,
                                "file": self.parse_file_object,
                                "ip-port": self.parse_ip_port_object,
                                "network-connection": self.parse_network_connection_object,
                                "network-socket": self.parse_network_socket_object,
                                "pe": self.store_pe,
                                "pe-section": self.store_pe,
                                "process": self.parse_process_object,
                                "registry-key": self.parse_regkey_object,
                                "url": self.parse_url_object,
                                "user-account": self.parse_user_account_object,
                                "whois": self.parse_whois,
                                "x509": self.parse_x509_object}

    def loadEvent(self):
        pathname = os.path.dirname(self.args[0])
        self.filename = "{}/tmp/{}".format(pathname, self.args[1])
        with open(self.filename, 'rt', encoding='utf-8') as f:
            self.json_event = json.loads(f.read())

    def generateEventPackages(self):
        try:
            return_type_to_package = {'json': ('to_json', {}),
                                      'xml': ('to_xml', {'include_namespaces': False, 'include_schemalocs': False, 'encoding': 'utf8'})}
            to_call, args = return_type_to_package[self.return_type]
            separator = None
            if self.json_event.get('response'):
                from misp_framing import stix_framing
                _, separator, _ = stix_framing(self.baseurl, self.orgname, self.return_type)
                stix_packages = [getattr(self.generate_package(event['Event']), to_call)(**args) for event in self.json_event['response']]
            else:
                stix_packages = [getattr(self.generate_package(self.json_event['Event']), to_call)(**args)]
            if self.return_type == 'xml':
                stix_packages = [s.decode() for s in stix_packages]
                stix_packages = ['\n            '.join(s.split('\n')[:-1]).replace('stix:STIX_Package', 'stix:Package') for s in stix_packages]
                stix_packages = ['            {}\n'.format(s) for s in stix_packages]
            else:
                stix_packages = ['{"package": %s}' % s for s in stix_packages]
            self.stix_package = separator.join(stix_packages) if len(stix_packages) > 1 else stix_packages[0]
            self.saveFile()
            print(json.dumps({'success': 1}))
        except Exception as e:
            print(json.dumps({'error': e.__str__()}))

{'fraction_non_alphanumeric': 0.09550703444102668, 'fraction_numerical': 0.004639201250567293, 'mean_word_length': 3.4727108705457828, 'pattern_counts': {'":': 40, '<': 3, '<?xml version=': 0, '>': 9, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 6}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1981522', 'n_tokens_mistral': 5782, 'n_tokens_neox': 5619, 'n_words': 1141}