Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def parse_maxminddb(db_path, ip):
try:
reader = maxminddb.open_database(db_path)
response = reader.get(ip)
reader.close()
return response
except FileNotFoundError:
print('DB not found')
print('SHUTTING DOWN')
exit()
except ValueError:
return False
def getPeerLocations(self, peers):
import maxminddb
db_path = self.getGeoipDb()
if not db_path:
self.log.debug("Not showing peer locations: no GeoIP database")
return False
geodb = maxminddb.open_database(db_path)
peers = list(peers.values())
# Place bars
peer_locations = []
placed = {} # Already placed bars here
for peer in peers:
# Height of bar
if peer.connection and peer.connection.last_ping_delay:
ping = round(peer.connection.last_ping_delay * 1000)
else:
ping = None
loc = self.getLoc(geodb, peer.ip)
if not loc:
continue
# Create position array
def __init__(self, outputqueue, config):
multiprocessing.Process.__init__(self)
# All the printing output should be sent to the outputqueue. The outputqueue is connected to another process called OutputProcess
self.outputqueue = outputqueue
# In case you need to read the slips.conf configuration file for your own configurations
self.config = config
# Start the DB
__database__.start(self.config)
# Open the maminddb offline db
try:
self.reader = maxminddb.open_database('modules/geoip/GeoLite2-Country.mmdb')
except:
self.print('Error opening the geolite2 db in ./GeoLite2-Country_20190402/GeoLite2-Country.mmdb. Please download it from https://geolite.maxmind.com/download/geoip/database/GeoLite2-Country.tar.gz. Please note it must be the MaxMind DB version.')
# To which channels do you wnat to subscribe? When a message arrives on the channel the module will wakeup
self.c1 = __database__.subscribe('new_ip')
# Set the timeout based on the platform. This is because the pyredis lib does not have officially recognized the timeout=None as it works in only macos and timeout=-1 as it only works in linux
if platform.system() == 'Darwin':
# macos
self.timeout = None
elif platform.system() == 'Linux':
self.timeout = None
else:
#??
self.timeout = None
def __init__(self, outputqueue, config):
multiprocessing.Process.__init__(self)
# All the printing output should be sent to the outputqueue. The outputqueue is connected to another process called OutputProcess
self.outputqueue = outputqueue
# In case you need to read the slips.conf configuration file for your own configurations
self.config = config
# Start the DB
__database__.start(self.config)
# Set the output queue of our database instance
__database__.setOutputQueue(self.outputqueue)
# Open the maminddb offline db
try:
self.reader = maxminddb.open_database('modules/asn/GeoLite2-ASN.mmdb')
except:
self.print('Error opening the geolite2 db in ./GeoLite2-Country_20190402/GeoLite2-Country.mmdb. Please download it from https://geolite.maxmind.com/download/geoip/database/GeoLite2-Country.tar.gz. Please note it must be the MaxMind DB version.')
# To which channels do you wnat to subscribe? When a message arrives on the channel the module will wakeup
self.c1 = __database__.subscribe('new_ip')
# Set the timeout based on the platform. This is because the pyredis lib does not have officially recognized the timeout=None as it works in only macos and timeout=-1 as it only works in linux
if platform.system() == 'Darwin':
# macos
self.timeout = None
elif platform.system() == 'Linux':
# linux
self.timeout = None
else:
#??
self.timeout = None
def get_maxmind_coords(ip):
try:
# yeah, this is hardcoded and probably shouldn't be
root_dir = "/usr/local/aclu/elections-api"
db_path = '%s/sources/maxmind/geolite2_city.mmdb' % root_dir
reader = maxminddb.open_database(db_path)
rsp = reader.get(ip)
return {
'ok': True,
'source': 'maxmind',
'maxmind_details': rsp,
'ip': ip,
'location': {
'latitude': rsp['location']['latitude'],
'longitude': rsp['location']['longitude']
}
}
except:
return {
'ok': False,
'error': 'Unable to lookup IP address with maxmind.'
}
def get_location(ip):
if ip is None:
return "Unknown"
with maxminddb.open_database(geolite2.geolite2_database()) as reader:
try:
match = reader.get(ip)
return match['country']['names']['en']
except Exception:
return "Unknown"
import math
import string
import re
import bisect
import socket
import cvmfs_api
import time
import threading
import cvmfs_globals
# TODO(jblomer): we should better separate the code that needs the maxminddb
# dependency from the code that doesn't
if not cvmfs_globals.CVMFS_UNITTESTS:
import maxminddb
gireader = maxminddb.open_database("/var/lib/cvmfs-server/geo/GeoLite2-City.mmdb")
positive_expire_secs = 60*60 # 1 hour
geo_cache_secs = 5*60 # 5 minutes
geo_cache_max_entries = 100000 # a ridiculously large but manageable number
# geo_cache entries are indexed by name and contain a tuple of
# (update time, geo record). Caching DNS lookups is more important
# than caching geo information but it's simpler and slightly more
# efficient to cache the geo information.
geo_cache = {}
# function came from http://www.johndcook.com/python_longitude_latitude.html
def distance_on_unit_sphere(lat1, long1, lat2, long2):