Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def wait_for_workflow_execution(workflow_execution, stack_resources, wait_seconds):
# disable unsigned HTTPS certificate warnings
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
workflow_id = workflow_execution['Id']
# It might take a few seconds for the step function to start, so we'll try to get the execution arn a few times
# before giving up
retries=0
# FIXME retry_limit = ceil(wait_seconds/5)
retry_limit = (wait_seconds//5)+1
while(retries
def test_verify_notification_results_in_indexing(self):
"""Agent indexes documents for which notifications are produced."""
# We disable certificate validation for testing.
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
to_index = [
"1712.04442", # flux capacitor
"1511.07473", # flux capacitor
"1604.04228", # flux capacitor
"1403.6219", # λ
"1404.3450", # $z_1$
"1703.09067", # $\lambda$
"1408.6682", # $\lambda$
"1607.05107", # Schröder
"1509.08727", # Schroder
"1710.01597", # Schroeder
"1708.07156", # w w
"1401.1012", # Wonmin Son
]
for document_id in to_index:
import socket
import requests
import urllib3
import jsonpickle
from typing import List
NETWORK_TYPE = "carrier_wifi_network"
admin_cert = (
"/var/opt/magma/certs/rest_admin.crt",
"/var/opt/magma/certs/rest_admin.key",
)
# Disable warnings about SSL verification since its a local VM
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
class AllowedGREPeers:
def __init__(self, ip: str, key: int):
self.ip = ip
self.key = key
class CarrierWiFiConfig:
def __init__(self, grePeers: List[AllowedGREPeers]):
self.allowed_gre_peers = grePeers
class NetworkDNSConfig:
def __init__(self, enable_caching: bool, local_ttl: int):
self.enable_caching = enable_caching
#!/usr/bin/python
import requests
import json
import pdb
import urllib3
import shutil
import time
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
class NessusRequest(object):
HEADERS = {}
def __init__(
self,
username,
password,
host,
verify=False,
proxies=None,
uuid=None,
folder_id=None,
policy_id=23,
#!/usr/bin/python
import re
import urllib3
import csv
import os
import sys
import time
import datetime
import numpy as np
from bs4 import BeautifulSoup
from dotenv import load_dotenv
load_dotenv()
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
api_key = os.getenv("NEWSAPI_APIKEY")
class news_Reuters:
def __init__(self):
fin = open('./input/tickerList.csv')
self.newsapi = NewsApiClient(api_key=api_key)
filterList = set()
try: # this is used when we restart a task
fList = open('./input/finished.reuters')
for l in fList:
filterList.add(l.strip())
except: pass
dateList = self.dateGenerator(1000) # look back on the past X days
for line in fin: # iterate all possible tickers
import requests
import json
import urllib3
from datetime import datetime
from aemet.constants import *
# Disable Insecure Request Warnings
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
class Prediccion:
def __init__(self, provincia, version, id, origen, elaborado, prediccion, nombre):
self.provincia = provincia
self.version = version
self.id = id
self.origen = origen
self.elaborado = elaborado
self.prediccion = prediccion
self.nombre = nombre
@staticmethod
def from_json(data, periodo):
prediccion = ''
if periodo == PERIODO_DIA:
import os
import requests
from urllib3.exceptions import InsecureRequestWarning
requests.packages.urllib3.disable_warnings(category=InsecureRequestWarning)
# Constants
BASE_URL = "https://netbox.lasthop.io/api/"
def main():
token = os.environ["NETBOX_TOKEN"]
http_headers = {}
http_headers["accept"] = "application/json; version=2.4;"
http_headers["Authorization"] = f"Token {token}"
# Retrieve all the devices
resp = requests.get(f"{BASE_URL}dcim/devices/", headers=http_headers, verify=False)
elif inven_resp.status_code == 400:
print("Inventory type %s not applicable for device with Id %s" % (inventory_type, device_id))
else:
print("Unable to retrieve inventory for device %s due to status code %s" % (device_id, inven_resp.status_code))
else:
print("Unable to retrieve details for device (%s) from %s" % (field, ip_address))
else:
print("No device data retrieved from %s" % (ip_address))
else:
print("Unable to create a session with appliance %s" % (ip_address))
except:
print("Unexpected error:", sys.exc_info())
if __name__ == '__main__':
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
PARSER = argparse.ArgumentParser(description=__doc__,
formatter_class=RawTextHelpFormatter)
PARSER.add_argument("--ip", "-i", required=True, help="OME Appliance IP")
PARSER.add_argument("--user", "-u", required=True,
help="Username for OME Appliance",
default="admin")
PARSER.add_argument("--password", "-p", required=True,
help="Password for OME Appliance")
PARSER.add_argument("--filterby", "-fby", required=True,
choices=('Id', 'Name','SvcTag'),
help="Filter by id/name/service tag")
PARSER.add_argument("--field", "-f", required=True,
help="Field to filter by (id/name/svc tag)")
PARSER.add_argument("--inventorytype", "-invtype", required=False,
choices=('cpus', 'os', 'disks', 'controllers', 'memory'),
help="Get inventory by cpus/os/disks/controllers,memory")
# Contact: d.machard@gmail.com
# Website: www.extensiveautomation.org
# -------------------------------------------------------------------
try:
from PyQt4.QtCore import ( QObject, pyqtSignal )
except ImportError:
from PyQt5.QtCore import ( QObject, pyqtSignal )
from Core import Settings
import requests
import urllib3
from urllib3.exceptions import InsecureRequestWarning
urllib3.disable_warnings(InsecureRequestWarning)
import xml.etree.ElementTree as ET
import os
import html
import time
from xml.sax.saxutils import escape
class RestHpAlmClient(QObject):
"""
Rest HP ALM client
"""
Error = pyqtSignal(object)
ConnectionOk = pyqtSignal()
TestsExported = pyqtSignal(list, dict)
ResultsExported = pyqtSignal(list, dict)
def __proxy_pool(self):
"""
Create Proxy connection pool
:raise ProxyRequestError
:return: urllib3.HTTPConnectionPool
"""
try:
self.__server = self.__cfg.proxy if True is self.__cfg.is_standalone_proxy else self.__get_random_proxy()
if self.__get_proxy_type(self.__server) == 'socks':
disable_warnings(InsecureRequestWarning)
if not hasattr(self, '__pm'):
package_module = importlib.import_module('urllib3.contrib.socks')
self.__pm = getattr(package_module, 'SOCKSProxyManager')
pool = self.__pm(self.__server,
num_pools=self.__cfg.threads,
timeout=Timeout(self.__cfg.timeout,
read=self.__cfg.timeout),
block=True)
else:
pool = ProxyManager(self.__server,
num_pools=self.__cfg.threads,
timeout=Timeout(self.__cfg.timeout, read=self.__cfg.timeout),
block=True)