Best Python code snippet using prospector_python
cli.py
Source:cli.py
1#!/usr/bin/env python2'''3MIT License4Copyright (c) 2019 Tenable Network Security, Inc.5Permission is hereby granted, free of charge, to any person obtaining a copy6of this software and associated documentation files (the "Software"), to deal7in the Software without restriction, including without limitation the rights8to use, copy, modify, merge, publish, distribute, sublicense, and/or sell9copies of the Software, and to permit persons to whom the Software is10furnished to do so, subject to the following conditions:11The above copyright notice and this permission notice shall be included in all12copies or substantial portions of the Software.13THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR14IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,15FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE16AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER17LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,18OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE19SOFTWARE.20'''21import click, logging, time, yaml, json, platform, sys, os, arrow22from tenable.io import TenableIO23from tenable.sc import TenableSC24from .config import base_config25from restfly.utils import dict_merge26from .jira import Jira27from .transform import Tio2Jira28from . import __version__29troubleshooting = '''30### Configuration File:31```yaml32{configfile}33```34### Debug Logs35```36{logging}37```38### Available IssueTypes39```yaml40{issuetypes}41```42'''43@click.command()44@click.option('--observed-since', '-s', envvar='SINCE', default=0,45 type=click.INT, help='The unix timestamp of the age threshold')46@click.option('--first-discovery', '-f', is_flag=True,47 help='Only add issues found for the first time within the age threshold')48@click.option('--setup-only', is_flag=True,49 help='Performs setup tasks and generates a config file.')50@click.option('--troubleshoot', is_flag=True,51 help='Outputs some basic troubleshooting data to file as an issue.')52@click.argument('configfile', default='config.yaml', type=click.File('r'))53def cli(configfile, observed_since, first_discovery=False, setup_only=False, troubleshoot=False):54 '''55 Tenable.io -> Jira Cloud Transformer & Ingester56 '''57 # Load the config, but ensure that any additional fields are additive to the58 # basic field set.59 config_from_file = yaml.safe_load(configfile)60 fields = config_from_file.pop('custom_fields', list())61 config = dict_merge(base_config(), config_from_file)62 config['fields'] = config['fields'] + fields63 sevprio = config['tenable'].get('severity_prioritization')64 if config['tenable'].get('tio_transform_tags'):65 attr_cache = config['tenable'].get('tio_asset_attr_cache', list())66 tag_attrs = config['tenable'].get('tio_transform_tags', list())67 config['tenable']['tio_asset_attr_cache'] = attr_cache + tag_attrs68 # Get the logging definition and define any defaults as need be.69 log = config.get('log', {})70 log_lvls = {'debug': 10, 'info': 20, 'warn': 30, 'error': 40}71 log['level'] = log_lvls[log.get('level', 'warn')]72 log['format'] = log.get('format',73 '%(asctime)-15s %(name)s %(levelname)s %(message)s')74 # Configure the root logging facility75 if troubleshoot:76 logging.basicConfig(77 level=logging.DEBUG,78 format=log['format'],79 filename='tenable_debug.log'80 )81 else:82 logging.basicConfig(**log)83 # Output some basic information detailing the config file used and the84 # python version & system arch.85 logging.info(f'Tenable2JiraCloud Version {__version__}')86 logging.info(f'Using configuration file {configfile.name}')87 uname = platform.uname()88 logging.info(('Running on Python'89 f'{".".join([str(i) for i in sys.version_info][0:3])}'90 f'{uname[0]}/{uname[-2]}'91 ))92 logging.debug(f'Severity Prioritization set to: {sevprio}')93 # instantiate the Jira object94 jira = Jira(95 'https://{}/rest/api/3'.format(config['jira']['address']),96 config['jira']['api_username'],97 config['jira']['api_token']98 )99 # Initiate the Tenable.io API model, the Ingester model, and start the100 # ingestion and data transformation.101 if config['tenable'].get('platform') == 'tenable.io':102 if not observed_since:103 # if no since field is supplied, then look in the config file to see104 # if an age was applied, if not, then use the default of 30 days.105 observed_since = arrow.now()\106 .shift(days=-config['tenable'].get('tio_age', 30))\107 .floor('day').timestamp()108 source = TenableIO(109 access_key=config['tenable'].get('access_key'),110 secret_key=config['tenable'].get('secret_key'),111 vendor='Tenable',112 product='JiraCloud',113 build=__version__114 )115 if int(source.session.details().get('permissions')) < 64:116 logging.error('API Keys tie to non-admin user.')117 elif config['tenable'].get('platform') == 'tenable.sc':118 source = TenableSC(119 config['tenable'].get('address'),120 port=int(config['tenable'].get('port', 443)),121 username=config['tenable'].get('username'),122 password=config['tenable'].get('password'),123 access_key=config['tenable'].get('access_key'),124 secret_key=config['tenable'].get('secret_key'),125 vendor='Tenable',126 product='JiraCloud',127 build=__version__128 )129 else:130 logging.error('No valid Tenable platform configuration defined.')131 exit(1)132 ingest = Tio2Jira(source, jira, config)133 if troubleshoot:134 # if the troubleshooting flag is set, then we will be collecting some135 # basic information and outputting it to the screen in a format that136 # Github issues would expect to format it all pretty. This should help137 # reduce the amount of time that is spent with back-and-forth debugging.138 try:139 ingest.ingest(int(observed_since), first_discovery)140 except:141 logging.exception('Caught the following Exception')142 # Some basic redaction of sensitive data, such as API Keys, Usernames,143 # Passwords, and hostnames.144 addr = config_from_file['jira']['address']145 sc_addr = 'NOTHING_TO_SEE_HERE_AT_ALL'146 config_from_file['jira']['address'] = '<REDACTED>'147 config_from_file['jira']['api_token'] = '<REDACTED>'148 config_from_file['jira']['api_username'] = '<REDACTED>'149 config_from_file['project']['leadAccountId'] = '<REDACTED>'150 if config_from_file['tenable'].get('address'):151 sc_addr = config_from_file['tenable']['address']152 config_from_file['tenable']['address'] = '<REDACTED>'153 if config_from_file['tenable'].get('access_key'):154 config_from_file['tenable']['access_key'] = '<REDACTED>'155 if config_from_file['tenable'].get('secret_key'):156 config_from_file['tenable']['secret_key'] = '<REDACTED>'157 if config_from_file['tenable'].get('username'):158 config_from_file['tenable']['username'] = '<REDACTED>'159 if config_from_file['tenable'].get('password'):160 config_from_file['tenable']['password'] = '<REDACTED>'161 output = troubleshooting.format(162 configfile=yaml.dump(config_from_file, default_flow_style=False),163 logging=open('tenable_debug.log').read() \164 .replace(addr, '<JIRA_CLOUD_HOST>') \165 .replace(sc_addr, '<TENABLE_SC_HOST>'),166 issuetypes='\n'.join(167 [168 '{id}: {name}'.format(**a)169 for a in jira.issue_types.list()170 if a.get('name').lower() in ['task', 'subtask', 'sub-task']171 ]172 )173 )174 print(output)175 print('\n'.join([176 '/-------------------------------NOTICE-----------------------------------\\',177 '| The output above is helpful for us to troubleshoot exactly what is |',178 '| happening within the code and offer a diagnosis for how to correct. |',179 '| Please note that while some basic redaction has already been performed |',180 '| that we ask you to review the information you\'re about to send and |',181 '| ensure that nothing deemed sensitive is transmitted. |',182 '| ---------------------------------------------------------------------- |',183 '| -- Copy of output saved to "issue_debug.md" |',184 '\\------------------------------------------------------------------------/'185 ]))186 with open('issue_debug.md', 'w') as reportfile:187 print(output, file=reportfile)188 os.remove('tenable_debug.log')189 elif not setup_only:190 # If we are expected to continually re-run the transformer, then we will191 # need to track the passage of time and run every X hours, where X is192 # defined by the user in the configuration.193 daemon = True194 last_run = int(observed_since)195 while daemon:196 since = last_run197 last_run = int(time.time())198 logging.info(f'Initiating ingest with observed_since={since}')199 ingest.ingest(since, first_discovery)200 if config.get('service', {}).get('interval', 0) > 0:201 sleeper = int(config['service']['interval']) * 3600202 logging.info(f'Sleeping for {sleeper} seconds')203 time.sleep(sleeper)204 else:205 daemon = False206 elif setup_only:207 # In setup-only mode, the ingest will not run, and instead a config file208 # will be generated that will have all of the JIRA identifiers baked in209 # and will also inform the integration to ignore the screen builder.210 # When using this config, if there are any changes to the code, then211 # this config will need to be re-generated.212 config['screen']['no_create'] = True213 logging.info('Set to setup-only. Will not run ingest.')214 logging.info('The following is the updated config file from the setup.')215 with open('generated_config.yaml', 'w') as outfile:216 outfile.write(yaml.dump(config, Dumper=yaml.Dumper))217 logging.info('Generated "generated_config.yaml" config file.')...
config_loader.py
Source:config_loader.py
1import os2import sys3BASE_DIR = None4import logging5def set_base_dir(value):6 global BASE_DIR7 BASE_DIR =value8def get_params(name):9 import sys10 extra_params = [x for x in sys.argv if x.__len__()>name.__len__()+2 and x[0:2+name.__len__()+1] == "--"+name+"="]11 if extra_params.__len__()==0:12 return None13 else:14 return extra_params[0].split('=')[1]15def load_config(file_name,none_params=False):16 global settings, x17 import imp18 settings = imp.new_module(file_name + ".settings")19 setattr(settings, "BASE_DIR", BASE_DIR)20 import json21 with open(BASE_DIR + os.sep + "configs" + os.sep + file_name + '.json') as f:22 config_from_file = json.load(f)23 if type(config_from_file["APPS"]) in [str,unicode]:24 with open(BASE_DIR + os.sep + "configs" + os.sep + config_from_file["APPS"] + '.json') as f2:25 APPS = json.load(f2)26 config_from_file.update({27 "APPS":APPS["APPS"]28 })29 for x in config_from_file.get("PACKAGES", []):30 try:31 sys.path.append(BASE_DIR + os.sep + x.replace("/", os.sep))32 print "add path '{0}'".format(BASE_DIR + os.sep + x.replace("/", os.sep))33 except Exception as ex:34 print "add path '{0}' error \n{1}".format(BASE_DIR + os.sep + x.replace("/", os.sep, ex.message))35 configs_items = []36 setattr(settings, "SECRET_KEY", config_from_file["SECRET_KEY"])37 for key in config_from_file.keys():38 try:39 if key == "LOGS":40 pass41 if key == "PACKAGES":42 pass43 elif key == "DB_BACK_END":44 pass45 elif key == "DB_API_CACHE":46 setattr(settings, key, config_from_file[key])47 elif key == "AUTHORIZATION_ENGINE":48 setattr(settings, key, config_from_file[key])49 elif key == "DB_AUTH":50 import quicky51 quicky.authorize.set_config(config_from_file[key])52 setattr(settings, key, config_from_file[key])53 elif key == "DB_LANGUAGE":54 import quicky55 quicky.language.set_config(config_from_file[key])56 setattr(settings, key, config_from_file[key])57 elif key == "DB_ENCRYPTOR_CACHE":58 from . import encryptor59 encryptor.set_config(config_from_file[key])60 setattr(settings, key, config_from_file[key])61 elif key == "DB_EXCEL_EXPORT_CONFIG":62 from . import language63 language.set_config(config_from_file[key])64 setattr(settings, key, config_from_file[key])65 elif key == "DB_TRACKING":66 import qtracking67 qtracking.set_config(config_from_file[key])68 setattr(settings,key,config_from_file[key])69 elif key=="JASPER":70 import qjasper71 qjasper.set_config(72 url=config_from_file[key]["URL"],73 user=config_from_file[key]["USER"],74 password=config_from_file[key]["PASSWORD"]75 )76 elif key == "APPS":77 pass78 else:79 setattr(settings, key, config_from_file[key])80 configs_items.append(key)81 print "load '{0}' with value {1}".format(key, config_from_file[key])82 except Exception as ex:83 txt_loaded_items = ""84 for x in configs_items:85 txt_loaded_items = txt_loaded_items + "\n\t\t" + x86 raise (Exception(87 "load '{0}.json' error, see details:\nloaded items:\n{1}\n error at item:\n '{2}'\n error message:\n{3}".format(88 file_name, txt_loaded_items, key, ex.message)))89 from django.conf.urls import url, include90 import importlib91 setattr(settings, "AUTHORIZATION_ENGINE", importlib.import_module(config_from_file["AUTHORIZATION_ENGINE"]))92 setattr(settings, "ROOT_URLCONF", 'apps')93 setattr(settings, "STATIC_URL", 'static/')94 setattr(settings, "STATIC_ROOT",95 os.path.join(*(BASE_DIR.split(os.path.sep) + ['apps/static', 'apps/app_main/static'])))96 LOGGING = {97 'version': 1,98 'disable_existing_loggers': False,99 'handlers': {100 'file': {101 'level': 'DEBUG',102 'class': 'logging.FileHandler',103 'filename': BASE_DIR + os.sep + config_from_file.get("LOGS", 'logs' + os.sep + 'debug.log'),104 },105 },106 'loggers': {107 'django': {108 'handlers': ['file'],109 'level': 'DEBUG',110 'propagate': False,111 },112 },113 }114 setattr(settings, "LOGGING", LOGGING)115 sys.modules.update({file_name: {"settings": settings}})116 sys.modules.update({file_name + ".settings": settings})117 os.environ.setdefault("DJANGO_SETTINGS_MODULE", file_name + ".settings")118 from . import api119 api.connect(config_from_file["DB_API_CACHE"])120 from . import backends121 backends.set_config(config_from_file["DB_BACK_END"])122 quicky.url.build_urls(settings.ROOT_URLCONF, [x for x in config_from_file["APPS"] if not x.get("disable", False)])123 from django.core.management import execute_from_command_line124 args = [x for x in sys.argv if x[0:2] != "--"]125 log = logging.getLogger(__file__)126 if not none_params:127 try:128 execute_from_command_line(args)129 except Exception as ex:130 log.debug(ex)131def start_app(name):132 if BASE_DIR == None:133 raise (Exception("It looks like you forgot call 'config_loader.set_base_dir' set root directory of app"))134 import imp135 settings = None136 file_name = get_params(name)...
archilles.py
Source:archilles.py
1#!/usr/bin/env python32import argparse3import requests4import validators5import yaml6from yaml import Loader7from urllib.parse import urlparse8from bs4 import BeautifulSoup9from bs4 import Comment10parser = argparse.ArgumentParser(description='The Archilles HTML Vulnerability Analyzer Version 1.0')11parser.add_argument('-v','--version', action='version', version='%(prog)s 1.0')12parser.add_argument('url',type=str, help='The URL of the HTML to analyze')13parser.add_argument('--config', help='path to configuration file')14parser.add_argument('-o', '--output', help='Report file output path')15args = parser.parse_args()16url = args.url17report = ''18config = {'forms': True, 'comments': True, 'passwords': True}19if(args.config):20 print('\nUsing config file: ' + args.config)21 config_file = open(args.config, 'r')22 config_from_file = yaml.load(config_file,Loader=Loader)23 if(config_from_file):24 config = { **config, **config_from_file }25if(validators.url(url)):26 result_html = requests.get(url).text27 parsed_html = BeautifulSoup(result_html, 'html.parser')28 forms = parsed_html.find_all('form')29 comments = parsed_html.find_all(string=lambda text:isinstance(text,Comment))30 password_inputs = parsed_html.find_all('input', { 'name' : 'password'})31 if(config['forms']):32 for form in forms:33 if (form.get('action').find('https') < 0) and urlparse(url).scheme != 'https':34 report += '[+] Form Issue: Insecure form action ' + form.get('action') + ' found in document\n'35 if(config['comments']):36 for comment in comments:37 if(comment.find('key: ') > -1):38 report += '[+] Comment Issue: Key is found in the HTML comments, please remove\n'39 if(config['passwords']):40 for password_input in password_inputs:41 if(password_input.get('type') != 'password'):42 report += '[+] Input Issue: Plaintext password input found. Please change to password type input\n'43else:44 print('Invalid URL. Please include full URL including scheme.')45if report == '':46 report = 'Nice Job! Your HTML document is secure'47else:48 header = 'Vulnerability Report is as follows:\n'49 header += '='* len(header) + '\n\n'50 report = header + report51print(report)52if(args.output):53 f = open(args.output, 'w')54 f.write(report)55 f.close...
Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!