This is my base script for using the Sumo Logic API to query logs and analyze data. This particular script finds hosts sending syslog data successfully through our firewall, looks who owns the netblock (they weren’t all internal!), and checks our configuration management database (cmdb) to see if we have a host registered with the destination IP address of the syslog traffic. <\/p>\n\n\n
\nimport requests\nfrom requests.auth import HTTPBasicAuth\nimport time\nfrom collections import defaultdict\nimport cx_Oracle\nimport pandas as pd\nimport ipaddress\nfrom datetime import datetime\nfrom ipwhois import IPWhois\nfrom ipwhois.exceptions import IPDefinedError\n\n# Import credentials from a config file\nfrom config import access_id, access_key, oracle_username, oracle_password\n\n# Initialize Oracle Client\ncx_Oracle.init_oracle_client(lib_dir=r"C:\\Oracle\\instantclient_21_15")\noracle_dsn = cx_Oracle.makedsn('cmdb_db.example.com', 1521, service_name='cmdb_db.example.com')\n\n# Function to query Oracle database\ndef query_oracle_cmdb(strIPAddress):\n with cx_Oracle.connect(user=oracle_username, password=oracle_password, dsn=oracle_dsn) as connection:\n cursor = connection.cursor()\n query = """\n SELECT HOSTNAME, FRIENDLYNAME, STATUS, COLLECTIONTIME, RETIREDBYDISPLAYNAME, \n RETIREDDATETIME, SERVERAPPSUPPORTTEAM, SERVERENVIRONMENT\n FROM NBIREPORT.CHERWELL_CMDBDATA_FULL\n WHERE IPADDRESS = :ipaddy\n """\n cursor.execute(query, [strIPAddress])\n result = cursor.fetchone()\n cursor.close()\n return result if result else ("",) * 8\n\n# Function to determine IP ownership\ndef get_ip_ownership(ip):\n # Define internal IP ranges\n internal_networks = [\n ipaddress.IPv4Network("10.0.0.0\/8"),\n ipaddress.IPv4Network("172.16.0.0\/12"),\n ipaddress.IPv4Network("192.168.0.0\/16")\n ]\n \n # Check if the IP is internal\n ip_obj = ipaddress.IPv4Address(ip)\n if any(ip_obj in network for network in internal_networks):\n return "INTERNAL"\n \n # For external IPs, use ipwhois to get ownership info\n try:\n obj = IPWhois(ip)\n result = obj.lookup_rdap(depth=1)\n ownership = result['network']['name']\n except IPDefinedError:\n ownership = "Reserved IP"\n except Exception as e:\n print(f"Error looking up IP {ip}: {e}")\n ownership = "UNKNOWN"\n \n return ownership\n\n# Base URL for Sumo Logic API\nbase_url = 'https:\/\/api.sumologic.com\/api\/v1'\n\n# Define the search query\nsearch_query = '''\n(dpt=514)\nAND _sourcecategory = "observe\/perimeter\/firewall\/logs"\n| where !(act = "deny")\n| where !(act = "timeout")\n| where !(act = "ip-conn")\n| where (proto=17 or proto=6)\n| count dst, act\n'''\n\n# Function to create and manage search jobs\ndef run_search_job(start_time, end_time):\n search_job_data = {\n 'query': search_query,\n 'from': start_time,\n 'to': end_time,\n 'timeZone': 'UTC'\n }\n\n # Create a search job\n search_job_url = f'{base_url}\/search\/jobs'\n response = requests.post(\n search_job_url,\n auth=HTTPBasicAuth(access_id, access_key),\n json=search_job_data\n )\n\n if response.status_code != 202:\n print('Error starting search job:', response.status_code, response.text)\n return None\n\n # Get the search job ID\n job_id = response.json()['id']\n print('Search Job ID:', job_id)\n\n # Poll for the search job status\n job_status_url = f'{search_job_url}\/{job_id}'\n while True:\n response = requests.get(job_status_url, auth=HTTPBasicAuth(access_id, access_key))\n status = response.json().get('state', None)\n print('Search Job Status:', status)\n if status in ['DONE GATHERING RESULTS', 'CANCELLED', 'FAILED']:\n break\n time.sleep(5) # Reasonable delay to prevent overwhelming the server\n\n return job_id if status == 'DONE GATHERING RESULTS' else None\n\n# Function to retrieve results of a search job\ndef retrieve_results(job_id):\n dst_counts = defaultdict(int)\n results_url = f'{base_url}\/search\/jobs\/{job_id}\/messages'\n offset = 0\n limit = 1000\n\n while True:\n params = {'offset': offset, 'limit': limit}\n try:\n response = requests.get(results_url, auth=HTTPBasicAuth(access_id, access_key), params=params, timeout=30)\n if response.status_code == 200:\n results = response.json()\n messages = results.get('messages', [])\n \n for message in messages:\n message_map = message['map']\n dst = message_map.get('dst')\n if dst:\n dst_counts[dst] += 1\n \n if len(messages) < limit:\n break\n\n offset += limit\n else:\n print('Error retrieving results:', response.status_code, response.text)\n break\n except requests.exceptions.RequestException as e:\n print(f'Error during request: {e}')\n time.sleep(5)\n continue\n\n return dst_counts\n\n# Main execution\nif __name__ == "__main__":\n # Prompt for the start date\n start_date_input = input("Enter the start date (YYYY-MM-DD): ")\n try:\n start_time = datetime.strptime(start_date_input, "%Y-%m-%d").strftime("%Y-%m-%dT00:00:00")\n except ValueError:\n print("Invalid date format. Please enter the date in YYYY-MM-DD format.")\n exit()\n\n # Use today's date as the end date\n end_time = datetime.now().strftime("%Y-%m-%dT00:00:00")\n\n # Create a search job\n job_id = run_search_job(start_time, end_time)\n if job_id:\n # Retrieve and process results\n dst_counts = retrieve_results(job_id)\n\n # Prepare data for Excel\n data_for_excel = []\n\n print("\\nDestination IP Counts and Oracle Data:")\n for dst, count in dst_counts.items():\n oracle_data = query_oracle_cmdb(dst)\n ownership = get_ip_ownership(dst)\n # Use only Oracle data columns\n combined_data = (dst, count, ownership) + oracle_data\n data_for_excel.append(combined_data)\n print(combined_data)\n\n # Create a DataFrame and write to Excel\n df = pd.DataFrame(data_for_excel, columns=[\n "IP Address", "Occurrence Count", "Ownership",\n "CMDB_Hostname", "CMDB_Friendly Name", "CMDB_Status", "CMDB_Collection Time", \n "CMDB_Retired By", "CMDB_Retired Date", "CMDB_Support Team", "CMDB_Environment"\n ])\n\n # Generate the filename with current date and time\n timestamp = datetime.now().strftime("%Y%m%d-%H%M")\n output_file = f"{timestamp}-sumo_oracle_data.xlsx"\n df.to_excel(output_file, index=False)\n print(f"\\nData written to {output_file}")\n else:\n print('Search job did not complete successfully.')\n<\/pre><\/div>","protected":false},"excerpt":{"rendered":"This is my base script for using the Sumo Logic API to query logs and analyze data. This particular script finds hosts sending syslog data successfully through our firewall, looks who owns the netblock (they weren’t all internal!), and checks our configuration management database (cmdb) to see if we have a host registered with the …<\/p>\n","protected":false},"author":1,"featured_media":0,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[2076],"tags":[664,2074,2075],"class_list":["post-11343","post","type-post","status-publish","format-standard","hentry","category-sumo-logic","tag-python","tag-sumo-logic","tag-sumo-logic-api"],"_links":{"self":[{"href":"https:\/\/www.rushworth.us\/lisa\/index.php?rest_route=\/wp\/v2\/posts\/11343","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/www.rushworth.us\/lisa\/index.php?rest_route=\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/www.rushworth.us\/lisa\/index.php?rest_route=\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/www.rushworth.us\/lisa\/index.php?rest_route=\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/www.rushworth.us\/lisa\/index.php?rest_route=%2Fwp%2Fv2%2Fcomments&post=11343"}],"version-history":[{"count":1,"href":"https:\/\/www.rushworth.us\/lisa\/index.php?rest_route=\/wp\/v2\/posts\/11343\/revisions"}],"predecessor-version":[{"id":11344,"href":"https:\/\/www.rushworth.us\/lisa\/index.php?rest_route=\/wp\/v2\/posts\/11343\/revisions\/11344"}],"wp:attachment":[{"href":"https:\/\/www.rushworth.us\/lisa\/index.php?rest_route=%2Fwp%2Fv2%2Fmedia&parent=11343"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/www.rushworth.us\/lisa\/index.php?rest_route=%2Fwp%2Fv2%2Fcategories&post=11343"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/www.rushworth.us\/lisa\/index.php?rest_route=%2Fwp%2Fv2%2Ftags&post=11343"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}