#!/usr/bin/env python # -*- coding: utf-8 -*- # # # # Licensed under the Apache License, Version 2.0 (the "License"): you may # not use this file except in compliance with the License. You may obtain # a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the # License for the specific language governing permissions and limitations # under the License. # # # Code modified from DNSTwist (https://github.com/elceef/dnstwist) # Thanks elceef! # # Changes made: # Just kept the DomainFuzz class and passing the domain to the fuzzer. Then added # the Splunk specific code around it # from __future__ import absolute_import, division, print_function, unicode_literals import sys import re import csv import time import os from splunklib.searchcommands import dispatch, GeneratingCommand, \ Configuration, Option, Boolean from splunk.clilib.bundle_paths import make_splunkhome_path class DomainFuzz(object): def __init__(self, domain): self.domain, self.tld = self.__domain_tld(domain) self.domains = [] self.qwerty = { '1': '2q', '2': '3wq1', '3': '4ew2', '4': '5re3', '5': '6tr4', '6': '7yt5', '7': '8uy6', '8': '9iu7', '9': '0oi8', '0': 'po9', 'q': '12wa', 'w': '3esaq2', 'e': '4rdsw3', 'r': '5tfde4', 't': '6ygfr5', 'y': '7uhgt6', 'u': '8ijhy7', 'i': '9okju8', 'o': '0plki9', 'p': 'lo0', 'a': 'qwsz', 's': 'edxzaw', 'd': 'rfcxse', 'f': 'tgvcdr', 'g': 'yhbvft', 'h': 'ujnbgy', 'j': 'ikmnhu', 'k': 'olmji', 'l': 'kop', 'z': 'asx', 'x': 'zsdc', 'c': 'xdfv', 'v': 'cfgb', 'b': 'vghn', 'n': 'bhjm', 'm': 'njk' } self.qwertz = { '1': '2q', '2': '3wq1', '3': '4ew2', '4': '5re3', '5': '6tr4', '6': '7zt5', '7': '8uz6', '8': '9iu7', '9': '0oi8', '0': 'po9', 'q': '12wa', 'w': '3esaq2', 'e': '4rdsw3', 'r': '5tfde4', 't': '6zgfr5', 'z': '7uhgt6', 'u': '8ijhz7', 'i': '9okju8', 'o': '0plki9', 'p': 'lo0', 'a': 'qwsy', 's': 'edxyaw', 'd': 'rfcxse', 'f': 'tgvcdr', 'g': 'zhbvft', 'h': 'ujnbgz', 'j': 'ikmnhu', 'k': 'olmji', 'l': 'kop', 'y': 'asx', 'x': 'ysdc', 'c': 'xdfv', 'v': 'cfgb', 'b': 'vghn', 'n': 'bhjm', 'm': 'njk' } self.azerty = { '1': '2a', '2': '3za1', '3': '4ez2', '4': '5re3', '5': '6tr4', '6': '7yt5', '7': '8uy6', '8': '9iu7', '9': '0oi8', '0': 'po9', 'a': '2zq1', 'z': '3esqa2', 'e': '4rdsz3', 'r': '5tfde4', 't': '6ygfr5', 'y': '7uhgt6', 'u': '8ijhy7', 'i': '9okju8', 'o': '0plki9', 'p': 'lo0m', 'q': 'zswa', 's': 'edxwqz', 'd': 'rfcxse', 'f': 'tgvcdr', 'g': 'yhbvft', 'h': 'ujnbgy', 'j': 'iknhu', 'k': 'olji', 'l': 'kopm', 'm': 'lp', 'w': 'sxq', 'x': 'zsdc', 'c': 'xdfv', 'v': 'cfgb', 'b': 'vghn', 'n': 'bhj' } self.keyboards = [self.qwerty, self.qwertz, self.azerty] def __domain_tld(self, domain): domain = domain.rsplit('.', 2) if len(domain) == 2: return domain[0], domain[1] return domain[0] + '.' + domain[1], domain[2] def __validate_domain(self, domain): if len(domain) == len(domain.encode('idna')) and domain != domain.encode('idna'): return False allowed = re.compile(b'(?=^.{4,253}$)(^((?!-)[a-zA-Z0-9-]{1,63}(?= 48 and o <= 57) or (o >= 97 and o <= 122) or o == 45: result.append(self.domain[:i] + b + self.domain[i+1:]) return result def __homoglyph(self): glyphs = { 'a': [u'à', u'á', u'â', u'ã', u'ä', u'å', u'ɑ', u'а', u'ạ', u'ǎ', u'ă', u'ȧ', u'ӓ'], 'b': ['d', 'lb', 'ib', u'ʙ', u'Ь', u'b̔', u'ɓ', u'Б'], 'c': [u'ϲ', u'с', u'ƈ', u'ċ', u'ć', u'ç'], 'd': ['b', 'cl', 'dl', 'di', u'ԁ', u'ժ', u'ɗ', u'đ'], 'e': [u'é', u'ê', u'ë', u'ē', u'ĕ', u'ě', u'ė', u'е', u'ẹ', u'ę', u'є', u'ϵ', u'ҽ'], 'f': [u'Ϝ', u'ƒ', u'Ғ'], 'g': ['q', u'ɢ', u'ɡ', u'Ԍ', u'Ԍ', u'ġ', u'ğ', u'ց', u'ǵ', u'ģ'], 'h': ['lh', 'ih', u'һ', u'հ', u'Ꮒ', u'н'], 'i': ['1', 'l', u'Ꭵ', u'í', u'ï', u'ı', u'ɩ', u'ι', u'ꙇ', u'ǐ', u'ĭ'], 'j': [u'ј', u'ʝ', u'ϳ', u'ɉ'], 'k': ['lk', 'ik', 'lc', u'κ', u'ⲕ', u'κ'], 'l': ['1', 'i', u'ɫ', u'ł'], 'm': ['n', 'nn', 'rn', 'rr', u'ṃ', u'ᴍ', u'м', u'ɱ'], 'n': ['m', 'r', u'ń'], 'o': ['0', u'Ο', u'ο', u'О', u'о', u'Օ', u'ȯ', u'ọ', u'ỏ', u'ơ', u'ó', u'ö', u'ӧ'], 'p': [u'ρ', u'р', u'ƿ', u'Ϸ', u'Þ'], 'q': ['g', u'զ', u'ԛ', u'գ', u'ʠ'], 'r': [u'ʀ', u'Г', u'ᴦ', u'ɼ', u'ɽ'], 's': [u'Ⴝ', u'Ꮪ', u'ʂ', u'ś', u'ѕ'], 't': [u'τ', u'т', u'ţ'], 'u': [u'μ', u'υ', u'Ս', u'ս', u'ц', u'ᴜ', u'ǔ', u'ŭ'], 'v': [u'ѵ', u'ν', u'v̇'], 'w': ['vv', u'ѡ', u'ա', u'ԝ'], 'x': [u'х', u'ҳ', u'ẋ'], 'y': [u'ʏ', u'γ', u'у', u'Ү', u'ý'], 'z': [u'ʐ', u'ż', u'ź', u'ʐ', u'ᴢ'] } result = [] for ws in range(0, len(self.domain)): for i in range(0, (len(self.domain)-ws)+1): win = self.domain[i:i+ws] j = 0 while j < ws: c = win[j] if c in glyphs: win_copy = win for g in glyphs[c]: win = win.replace(c, g) result.append(self.domain[:i] + win + self.domain[i+ws:]) win = win_copy j += 1 return list(set(result)) def __hyphenation(self): result = [] for i in range(1, len(self.domain)): result.append(self.domain[:i] + '-' + self.domain[i:]) return result def __insertion(self): result = [] for i in range(1, len(self.domain)-1): for keys in self.keyboards: if self.domain[i] in keys: for c in keys[self.domain[i]]: result.append(self.domain[:i] + c + self.domain[i] + self.domain[i+1:]) result.append(self.domain[:i] + self.domain[i] + c + self.domain[i+1:]) return list(set(result)) def __omission(self): result = [] for i in range(0, len(self.domain)): result.append(self.domain[:i] + self.domain[i+1:]) n = re.sub(r'(.)\1+', r'\1', self.domain) if n not in result and n != self.domain: result.append(n) return list(set(result)) def __repetition(self): result = [] for i in range(0, len(self.domain)): if self.domain[i].isalpha(): result.append(self.domain[:i] + self.domain[i] + self.domain[i] + self.domain[i+1:]) return list(set(result)) def __replacement(self): result = [] for i in range(0, len(self.domain)): for keys in self.keyboards: if self.domain[i] in keys: for c in keys[self.domain[i]]: result.append(self.domain[:i] + c + self.domain[i+1:]) return list(set(result)) def __subdomain(self): result = [] for i in range(1, len(self.domain)): if self.domain[i] not in ['-', '.'] and self.domain[i-1] not in ['-', '.']: result.append(self.domain[:i] + '.' + self.domain[i:]) return result def __transposition(self): result = [] for i in range(0, len(self.domain)-1): if self.domain[i+1] != self.domain[i]: result.append(self.domain[:i] + self.domain[i+1] + self.domain[i] + self.domain[i+2:]) return result def __vowel_swap(self): vowels = 'aeiou' result = [] for i in range(0, len(self.domain)): for vowel in vowels: if self.domain[i] in vowels: result.append(self.domain[:i] + vowel + self.domain[i+1:]) return list(set(result)) def __addition(self): result = [] for i in range(97, 123): result.append(self.domain + chr(i)) return result def generate(self): self.domains.append({'fuzzer': 'Original*', 'domain-name': self.domain + '.' + self.tld}) for domain in self.__addition(): self.domains.append({'fuzzer': 'Addition', 'domain-name': domain + '.' + self.tld}) for domain in self.__bitsquatting(): self.domains.append({'fuzzer': 'Bitsquatting', 'domain-name': domain + '.' + self.tld}) for domain in self.__homoglyph(): self.domains.append({'fuzzer': 'Homoglyph', 'domain-name': domain + '.' + self.tld}) for domain in self.__hyphenation(): self.domains.append({'fuzzer': 'Hyphenation', 'domain-name': domain + '.' + self.tld}) for domain in self.__insertion(): self.domains.append({'fuzzer': 'Insertion', 'domain-name': domain + '.' + self.tld}) for domain in self.__omission(): self.domains.append({'fuzzer': 'Omission', 'domain-name': domain + '.' + self.tld}) for domain in self.__repetition(): self.domains.append({'fuzzer': 'Repetition', 'domain-name': domain + '.' + self.tld}) for domain in self.__replacement(): self.domains.append({'fuzzer': 'Replacement', 'domain-name': domain + '.' + self.tld}) for domain in self.__subdomain(): self.domains.append({'fuzzer': 'Subdomain', 'domain-name': domain + '.' + self.tld}) for domain in self.__transposition(): self.domains.append({'fuzzer': 'Transposition', 'domain-name': domain + '.' + self.tld}) for domain in self.__vowel_swap(): self.domains.append({'fuzzer': 'Vowel-swap', 'domain-name': domain + '.' + self.tld}) if not self.domain.startswith('www.'): self.domains.append({'fuzzer': 'Various', 'domain-name': 'ww' + self.domain + '.' + self.tld}) self.domains.append({'fuzzer': 'Various', 'domain-name': 'www' + self.domain + '.' + self.tld}) self.domains.append({'fuzzer': 'Various', 'domain-name': 'www-' + self.domain + '.' + self.tld}) if '.' in self.tld: self.domains.append({'fuzzer': 'Various', 'domain-name': self.domain + '.' + self.tld.split('.')[-1]}) self.domains.append({'fuzzer': 'Various', 'domain-name': self.domain + self.tld}) if '.' not in self.tld: self.domains.append({'fuzzer': 'Various', 'domain-name': self.domain + self.tld + '.' + self.tld}) if self.tld != 'com' and '.' not in self.tld: self.domains.append({'fuzzer': 'Various', 'domain-name': self.domain + '-' + self.tld + '.com'}) self.__filter_domains() @Configuration(distributed=True) class DnsTwistCommand(GeneratingCommand): domainlist_file_name = Option(doc=''' **Syntax:** **domainlist=**** **Description:** CSV file from which repeated random samples will be drawn ''', name='domainlist', require=False) populate_from_cim = Option(doc=''' **Syntax: populate_cim= **Description:** When `true`, populates Splunk_SA_CIM lookups cim_corporate_email_domains.csv and cim_corporate_web_domains.csv with dnstwisted domains. Defaults to `false`. ''', name='populate_from_cim', default=False, validate=Boolean()) domain = Option(doc=''' **Syntax:** **domain=**** **Description:** Domain to DNS generated twisted entries for. ''', name='domain', require=False, default='') def generate(self): event_count = 0 csv_file_names = [] if self.populate_from_cim: csv_file_names.append(make_splunkhome_path([ 'etc', 'apps', 'Splunk_SA_CIM', 'lookups', 'cim_corporate_email_domains.csv'])) csv_file_names.append(make_splunkhome_path([ 'etc', 'apps', 'Splunk_SA_CIM', 'lookups', 'cim_corporate_web_domains.csv'])) # Make sure we just get the base file name from file. In case there was some directory traversal going on. if self.domainlist_file_name: sanitized_file_name = os.path.basename(self.domainlist_file_name) lookup_path = make_splunkhome_path(['etc', 'apps', 'DA-ESS-ContentUpdate', 'lookups', sanitized_file_name]) # Make sure there really isn't any directory traversal going on. valid_path = True if "../" in lookup_path: valid_path = False # Make sure the path that is created by adding the file name to the path is the same as the # absolute path if lookup_path != os.path.abspath(lookup_path): valid_path = False if valid_path: csv_file_names.append(lookup_path) domains_to_twist = [] for csv_file_name in csv_file_names: if os.path.exists(csv_file_name): # this is nasty but works .. please forgive me if sys.version_info >= (3, 0): csv_file = open(csv_file_name, "r", newline='') else: csv_file = open(csv_file_name, "r") for input_domain in csv.DictReader(csv_file): if input_domain['domain'] not in domains_to_twist: domains_to_twist.append(input_domain['domain']) # if a single domain is passed lets just calculate that if self.domain != '': domains_to_twist = [] domains_to_twist.append(self.domain) for domain_to_twist in domains_to_twist: domain_to_twist = domain_to_twist.lstrip('*') dfuzz = DomainFuzz(domain_to_twist) dfuzz.generate() domains = dfuzz.domains for domain in domains: # We don't want to keep the original domain if domain['domain-name'] in domain_to_twist: continue event_count += 1 yield { '_time': time.time(), 'event_no': event_count, '_raw': domain['domain-name'], 'domain': '*'+domain['domain-name']+'*', 'original_domain': domain_to_twist } def __init__(self): super(DnsTwistCommand, self).__init__() dispatch(DnsTwistCommand, sys.argv, sys.stdin, sys.stdout, __name__)