feat(yaml): add oob testing support

pocsuite3/lib/core/register.py

@@ -9,7 +9,6 @@
 from pocsuite3.lib.core.data import kb
 from pocsuite3.lib.core.data import logger
 from pocsuite3.lib.core.settings import POC_IMPORTDICT
-from pocsuite3.lib.yaml.nuclei import Nuclei
 
 
 class PocLoader(Loader):
@@ -72,6 +71,7 @@ def exec_module(self, module):
 
  # convert yaml template to pocsuite3 poc script
  if filename.endswith('.yaml') and re.search(r'matchers:\s+-', poc_code):
+ from pocsuite3.lib.yaml.nuclei import Nuclei
  poc_code = str(Nuclei(poc_code))
 
  self.check_requires(poc_code)

pocsuite3/lib/yaml/nuclei/__init__.py

@@ -77,6 +77,7 @@ def __init__(self, template, target=''):
  self.execute_options = OrderedDict()
  self.execute_options['stop_at_first_match'] = self.template.stop_at_first_match
  self.execute_options['variables'] = self.template.variables
+ self.execute_options['interactsh'] = None
 
  self.requests = self.template.requests
 
@@ -103,15 +104,16 @@ def execute_request(self, request: HttpRequest) -> dict:
  logger.debug(dump.dump_all(response).decode('utf-8'))
  except Exception as e1:
  logger.debug(str(e1))
- continue
- match_res = HttpMatch(request, response)
+ response = None
+ match_res = HttpMatch(request, response, self.execute_options['interactsh'])
  extractor_res = HttpExtract(request, response)
  if match_res and extractor_res:
  match_res = str(dict(extractor_res[0]))
  if match_res and request.stop_at_first_match:
  return match_res
  results.append(match_res)
- response.close()
+ if response:
+ response.close()
  except Exception as e:
  logger.debug(str(e))
  return results and any(results)
@@ -152,11 +154,9 @@ def execute_template(self):
  with automatic Request correlation built in. It's as easy as writing {{interactsh-url}} anywhere in the request.
  """
  if '{{interactsh-url}}' in self.yaml_template or '§interactsh-url§' in self.yaml_template:
- from pocsuite3.modules.interactsh import Interactsh
- ish = Interactsh()
- ish_url, ish_flag = ish.build_request(method='')
- self.dynamic_values['interactsh-url'] = ish_url
- self.execute_options['interactsh_client'] = ish
+ from pocsuite3.lib.yaml.nuclei.protocols.common.interactsh import InteractshClient
+ self.execute_options['interactsh'] = InteractshClient()
+ self.dynamic_values['interactsh-url'] = self.execute_options['interactsh'].client.domain
 
  results = []
  for request in self.requests:

pocsuite3/lib/yaml/nuclei/protocols/common/interactsh/__init__.py

@@ -0,0 +1,19 @@
+from dataclasses import dataclass, field
+from pocsuite3.modules.interactsh import Interactsh
+from pocsuite3.lib.core.log import LOGGER as logger
+
+
+@dataclass
+class InteractshClient:
+ client: Interactsh = field(default_factory=Interactsh)
+ interactsh_protocol: list = field(default_factory=list)
+ interactsh_request: list = field(default_factory=list)
+ interactsh_response: list = field(default_factory=list)
+
+ def poll(self) -> None:
+ results = self.client.poll()
+ for result in results:
+ logger.debug(result)
+ self.interactsh_protocol.append(result['protocol'])
+ self.interactsh_request.append(result['raw-request'])
+ self.interactsh_response.append(result['raw-response'])

pocsuite3/lib/yaml/nuclei/protocols/http/__init__.py

@@ -112,30 +112,47 @@ class HttpRequest:
  digest_password: str = ''
 
 
-def getMatchPart(part: str, response: requests.Response, return_bytes: bool = False) -> str:
+def getMatchPart(part: str, response: requests.Response, interactsh, return_bytes: bool = False) -> str:
  result = b''
- headers = '\n'.join(f'{k}: {v}' for k, v in response.headers.items()).encode('utf-8')
+ headers = b''
+ body = b''
+ if isinstance(response, requests.Response):
+ headers = '\n'.join(f'{k}: {v}' for k, v in response.headers.items()).encode('utf-8')
+ body = response.content
 
  if part == 'all':
- result = headers + b'\n\n' + response.content
+ result = headers + b'\n\n' + body
  elif part in ['', 'body']:
- result = response.content
+ result = body
  elif part in ['header', 'all_headers']:
  result = headers
-
- return result if return_bytes else result.decode('utf-8')
-
-
-def HttpMatch(request: HttpRequest, response: requests.Response):
+ elif part == 'interactsh_protocol':
+ interactsh.poll()
+ result = '\n'.join(interactsh.interactsh_protocol)
+ elif part == 'interactsh_request':
+ interactsh.poll()
+ result = '\n'.join(interactsh.interactsh_request)
+ elif part == 'interactsh_response':
+ interactsh.poll()
+ result = '\n'.join(interactsh.interactsh_response)
+
+ if return_bytes and not isinstance(result, bytes):
+ result = result.encode()
+ elif not return_bytes and isinstance(result, bytes):
+ result = result.decode()
+ return result
+
+
+def HttpMatch(request: HttpRequest, response: requests.Response, interactsh):
  matchers = request.matchers
  matchers_result = []
 
  for i, matcher in enumerate(matchers):
  matcher_res = False
- item = getMatchPart(matcher.part, response, return_bytes=matcher.type == MatcherType.BinaryMatcher)
+ item = getMatchPart(matcher.part, response, interactsh, return_bytes=matcher.type == MatcherType.BinaryMatcher)
 
  if matcher.type == MatcherType.StatusMatcher:
- matcher_res = MatchStatusCode(matcher, response.status_code)
+ matcher_res = MatchStatusCode(matcher, response.status_code if response else 0)
  logger.debug(f'matcher: {matcher}, result: {matcher_res}')
 
  elif matcher.type == MatcherType.SizeMatcher:
@@ -187,7 +204,7 @@ def HttpExtract(request: HttpRequest, response: requests.Response):
  res = ExtractRegex(extractor, item)
  logger.debug(f'extractor: {extractor}, result: {res}')
  elif extractor.type == ExtractorType.KValExtractor:
- res = ExtractKval(extractor, response.headers)
+ res = ExtractKval(extractor, response.headers if response else {})
  logger.debug(f'extractor: {extractor}, result: {res}')
  elif extractor.type == ExtractorType.XPathExtractor:
  res = ExtractXPath(extractor, item)

pocsuite3/modules/interactsh/__init__.py

@@ -5,12 +5,16 @@
 import json
 import random
 import time
-from uuid import uuid4
 from base64 import b64encode
+from uuid import uuid4
+
 from Cryptodome.Cipher import AES, PKCS1_OAEP
-from Cryptodome.PublicKey import RSA
 from Cryptodome.Hash import SHA256
-from pocsuite3.api import requests, logger, random_str, conf
+from Cryptodome.PublicKey import RSA
+
+from pocsuite3.lib.core.data import conf, logger
+from pocsuite3.lib.request import requests
+from pocsuite3.lib.utils import random_str
 
 
 class Interactsh:
@@ -53,12 +57,12 @@ def register(self):
  msg = f"[PLUGIN] Interactsh: Can not initiate {self.server} DNS callback client"
  try:
  res = self.session.post(
- f"https:https://{self.server}/register", headers=self.headers, json=data, verify=False)
+ f"http:https://{self.server}/register", headers=self.headers, json=data, verify=False)
  if res.status_code == 401:
  logger.error("[PLUGIN] Interactsh: auth error")
  elif 'success' not in res.text:
  logger.error(msg)
- except requests.exceptions.RequestException:
+ except requests.RequestException:
  logger.error(msg)
 
  def poll(self):
@@ -67,7 +71,7 @@ def poll(self):
  while count:
 
  try:
- url = f"https:https://{self.server}/poll?id={self.correlation_id}&secret={self.secret}"
+ url = f"http:https://{self.server}/poll?id={self.correlation_id}&secret={self.secret}"
  res = self.session.get(url, headers=self.headers, verify=False).json()
  aes_key, data_list = res['aes_key'], res['data']
  for i in data_list: