+++ /dev/null
-#!/usr/bin/python
-
-import ipaddress
-import logging
-import pycares
-import re
-import socket
-import tornado.gen
-import tornado.platform.caresresolver
-
-from . import countries
-
-from .decorators import *
-from .misc import Object
-
-# These lists are used to block access to the webapp
-BLOCKLISTS = (
- "sbl.spamhaus.org",
- "xbl.spamhaus.org",
-)
-
-BLACKLISTS = (
- "b.barracudacentral.org",
- "bl.spamcop.net",
- "bl.blocklist.de",
- "cbl.abuseat.org",
- "dnsbl-1.uceprotect.net",
- "dnsbl-2.uceprotect.net",
- "dnsbl-3.uceprotect.net",
- "dnsbl.abuse.ch",
- "ix.dnsbl.manitu.net",
- "pbl.spamhaus.org",
- "sbl.spamhaus.org",
- "xbl.spamhaus.org",
- "zen.spamhaus.org",
-)
-
-class Resolver(tornado.platform.caresresolver.CaresResolver):
- def initialize(self, **kwargs):
- super().initialize()
-
- # Overwrite Channel
- self.channel = pycares.Channel(sock_state_cb=self._sock_state_cb, **kwargs)
-
- async def query(self, name, type=pycares.QUERY_TYPE_A):
- # Create a new Future
- fut = tornado.gen.Future()
-
- # Perform the query
- self.channel.query(name, type, lambda result, error: fut.set_result((result, error)))
-
- # Wait for the response
- result, error = await fut
-
- # Handle any errors
- if error:
- # NXDOMAIN
- if error == pycares.errno.ARES_ENOTFOUND:
- return
-
- # Ignore responses with no data
- elif error == pycares.errno.ARES_ENODATA:
- return
-
- raise IOError(
- "C-Ares returned error %s: %s while resolving %s"
- % (error, pycares.errno.strerror(error), name)
- )
-
- # Return the result
- return result
-
-
-class GeoIP(Object):
- @lazy_property
- def resolver(self):
- return Resolver(tries=2, timeout=2, domains=[])
-
- def lookup(self, address):
- return Address(self.backend, address)
-
- def guess_address_family(self, addr):
- if ":" in addr:
- return 6
-
- return 4
-
- def get_country(self, addr):
- ret = self.get_all(addr)
-
- if ret:
- return ret.country
-
- def get_location(self, addr):
- query = "SELECT * FROM geoip \
- WHERE %s BETWEEN start_ip AND end_ip LIMIT 1"
-
- return self.db.get(query, addr)
-
- def get_asn(self, addr):
- query = "SELECT asn FROM geoip_asn \
- WHERE %s BETWEEN start_ip AND end_ip LIMIT 1"
-
- ret = self.db.get(query, addr)
-
- if ret:
- return ret.asn
-
- def get_all(self, addr):
- location = self.get_location(addr)
-
- if location:
- location["asn"] = self.get_asn(addr)
-
- return location
-
- _countries = {
- "A1" : "Anonymous Proxy",
- "A2" : "Satellite Provider",
- "AP" : "Asia/Pacific Region",
- "EU" : "Europe",
- }
-
- def get_country_name(self, code):
- return countries.get_name(code)
-
- async def test_blacklist(self, address):
- address = self.lookup(address)
-
- # Determne blacklist status
- status = await address.is_blacklisted()
-
- print("Blacklist status for %s: %s" % (address, status))
-
-
-class Address(Object):
- def init(self, address):
- self.address = ipaddress.ip_address(address)
-
- def __str__(self):
- return "%s" % self.address
-
- @property
- def family(self):
- if isinstance(self.address, ipaddress.IPv6Address):
- return socket.AF_INET6
- elif isinstance(self.address, ipaddress.IPv4Address):
- return socket.AF_INET
-
- # Blacklist
-
- def _make_blacklist_rr(self, blacklist):
- if self.family == socket.AF_INET6:
- octets = list(self.address.exploded.replace(":", ""))
- elif self.family == socket.AF_INET:
- octets = str(self.address).split(".")
- else:
- raise NotImplementedError("Unknown IP protocol")
-
- # Reverse the list
- octets.reverse()
-
- # Append suffix
- octets.append(blacklist)
-
- return ".".join(octets)
-
- async def _resolve_blacklist(self, blacklist):
- return_code = None
-
- # Get resource record name
- rr = self._make_blacklist_rr(blacklist)
-
- # Get query type from IP protocol version
- if self.family == socket.AF_INET6:
- type = pycares.QUERY_TYPE_AAAA
- elif self.family == socket.AF_INET:
- type = pycares.QUERY_TYPE_A
- else:
- raise NotImplementedError("Unknown IP protocol")
-
- # Run query
- try:
- res = await self.backend.geoip.resolver.query(rr, type=type)
- except IOError as e:
- logging.warning(e)
-
- return return_code, "%s" % e
-
- # Not found
- if not res:
- logging.debug("%s is not blacklisted on %s" % (self, blacklist))
- return return_code, None
-
- # Extract return code from DNS response
- for row in res:
- return_code = row.host
- break
-
- # If the IP address is on a blacklist, we will try to fetch the TXT record
- reason = await self.backend.geoip.resolver.query(rr, type=pycares.QUERY_TYPE_TXT)
-
- # Log result
- logging.debug("%s is blacklisted on %s: %s" % (self, blacklist, reason or "N/A"))
-
- # Take the first reason
- if reason:
- for i in reason:
- return return_code, i.text
-
- # Blocked, but no reason
- return return_code, None
-
- async def get_blacklists(self):
- blacklists = { bl : self._resolve_blacklist(bl) for bl in BLACKLISTS }
-
- return blacklists
-
- async def is_blacklisted(self):
- logging.debug("Checking if %s is blacklisted..." % self)
-
- # Perform checks
- blacklists = { bl : self._resolve_blacklist(bl) for bl in BLOCKLISTS }
-
- # If we are blacklisted on one list, this one is screwed
- for bl in blacklists:
- code, message = await blacklists[bl]
-
- logging.debug("Response from %s is: %s (%s)" % (bl, code, message))
-
- # Exclude matches on SBLCSS
- if bl == "sbl.spamhaus.org" and code == "127.0.0.3":
- continue
-
- # Consider the host blocked for any non-zero return code
- if code:
- return True
projects / ipfire.org.git / blobdiff