from fba.helpers import blacklist
from fba.helpers import cache
from fba.helpers import config
+from fba.helpers import domain as domain_helper
from fba.http import federation
from fba.http import network
# written to database. Both arrays must be filled at the same time or else
# update_data() will fail
_pending = {
- # Detection mode: 'AUTO_DISCOVERY', 'STATIC_CHECKS' or 'GENERATOR'
+ # Detection mode
# NULL means all detection methods have failed (maybe still reachable instance)
"detection_mode" : {},
# Found nodeinfo URL
"last_status_code" : {},
# Last error details
"last_error_details" : {},
+ # Wether obfuscation has been used
+ "has_obfuscation" : {},
}
def _set_data(key: str, domain: str, value: any):
- logger.debug(f"key='{key}',domain='{domain}',value[]='{type(value)}' - CALLED!")
+ logger.debug("key='%s',domain='%s',value[]='%s' - CALLED!", key, domain, type(value))
+ domain_helper.raise_on(domain)
if not isinstance(key, str):
raise ValueError("Parameter key[]='{type(key)}' is not 'str'")
elif key == "":
raise ValueError("Parameter 'key' is empty")
- elif not isinstance(domain, str):
- raise ValueError(f"Parameter domain[]='{type(domain)}' is not 'str'")
- elif domain == "":
- raise ValueError("Parameter 'domain' is empty")
- elif domain.lower() != domain:
- raise ValueError(f"Parameter domain='{domain}' must be all lower-case")
- elif not validators.domain(domain.split("/")[0]):
- raise ValueError(f"domain='{domain}' is not a valid domain")
- elif domain.endswith(".arpa"):
- raise ValueError(f"domain='{domain}' is a domain for reversed IP addresses, please don't crawl them!")
- elif domain.endswith(".tld"):
- raise ValueError(f"domain='{domain}' is a fake domain, please don't crawl them!")
elif not key in _pending:
raise ValueError(f"key='{key}' not found in _pending")
elif not utils.is_primitive(value):
logger.debug("EXIT!")
def has_pending(domain: str) -> bool:
- logger.debug("domain(%d)='%s' - CALLED!", len(domain), domain)
- if not isinstance(domain, str):
- raise ValueError(f"Parameter domain[]='{type(domain)}' is not 'str'")
- elif domain == "":
- raise ValueError("Parameter 'domain' is empty")
- elif domain.lower() != domain:
- raise ValueError(f"Parameter domain='{domain}' must be all lower-case")
- elif not validators.domain(domain.split("/")[0]):
- raise ValueError(f"domain='{domain}' is not a valid domain")
- elif domain.endswith(".arpa"):
- raise ValueError(f"domain='{domain}' is a domain for reversed IP addresses, please don't crawl them!")
- elif domain.endswith(".tld"):
- raise ValueError(f"domain='{domain}' is a fake domain, please don't crawl them!")
+ logger.debug("domain='%s' - CALLED!", domain)
+ domain_helper.raise_on(domain)
has = False
for key in _pending:
- logger.debug(f"key='{key}',domain='{domain}',_pending[key]()='{len(_pending[key])}'")
+ logger.debug("key='%s',domain='%s',_pending[key]()=%d", key, domain, len(_pending[key]))
if domain in _pending[key]:
has = True
break
- logger.debug(f"has='{has}' - EXIT!")
+ logger.debug("has='%s' - EXIT!", has)
return has
def update_data(domain: str):
- logger.debug("domain(%d)='%s' - CALLED!", len(domain), domain)
- if not isinstance(domain, str):
- raise ValueError(f"Parameter domain[]='{type(domain)}' is not 'str'")
- elif domain == "":
- raise ValueError("Parameter 'domain' is empty")
- elif domain.lower() != domain:
- raise ValueError(f"Parameter domain='{domain}' must be all lower-case")
- elif not validators.domain(domain.split("/")[0]):
- raise ValueError(f"domain='{domain}' is not a valid domain")
- elif domain.endswith(".arpa"):
- raise ValueError(f"domain='{domain}' is a domain for reversed IP addresses, please don't crawl them!")
- elif domain.endswith(".tld"):
- raise ValueError(f"domain='{domain}' is a fake domain, please don't crawl them!")
- elif not has_pending(domain):
+ logger.debug("domain='%s' - CALLED!", domain)
+ domain_helper.raise_on(domain)
+ if not has_pending(domain):
raise Exception(f"domain='{domain}' has no pending instance data, but function invoked")
elif not is_registered(domain):
raise Exception(f"domain='{domain}' cannot be updated while not being registered")
- logger.debug(f"Updating instance data for domain='{domain}' ...")
+ logger.debug("Updating instance data for domain='%s' ...", domain)
sql_string = ""
fields = list()
for key in _pending:
- logger.debug("key:", key)
+ logger.debug("Checking key='%s',domain='%s'", key, domain)
if domain in _pending[key]:
- logger.debug(f"Adding '{_pending[key][domain]}' for key='{key}' ...")
+ logger.debug("Adding '%s' for key='%s' ...", _pending[key][domain], key)
fields.append(_pending[key][domain])
sql_string += f" {key} = ?,"
- logger.debug(f"sql_string()={len(sql_string)}")
+ logger.debug("sql_string()=%d", len(sql_string))
if sql_string == "":
raise ValueError(f"No fields have been set, but method invoked, domain='{domain}'")
# For WHERE statement
fields.append(domain)
- logger.debug(f"sql_string='{sql_string}',fields()={len(fields)}")
+ logger.debug("sql_string='%s',fields()=%d", sql_string, len(fields))
sql_string = "UPDATE instances SET" + sql_string + " last_updated = ? WHERE domain = ? LIMIT 1"
- logger.debug("sql_string:", sql_string)
- logger.debug("Executing SQL:", sql_string)
+ logger.debug("Executing SQL: '%s'", sql_string)
database.cursor.execute(sql_string, fields)
- logger.debug(f"Success! (rowcount={database.cursor.rowcount })")
+ logger.debug("rowcount=%d", database.cursor.rowcount)
if database.cursor.rowcount == 0:
raise Exception(f"Did not update any rows: domain='{domain}',fields()={len(fields)}")
logger.debug("Invoking commit() ...")
database.connection.commit()
- logger.debug(f"Deleting _pending for domain='{domain}'")
+ logger.debug("Deleting _pending for domain='%s'", domain)
for key in _pending:
- logger.debug(f"domain='{domain}',key='{key}'")
+ logger.debug("domain='%s',key='%s'", domain, key)
if domain in _pending[key]:
+ logger.debug("Deleting key='%s',domain='%s' ...", key, domain)
del _pending[key][domain]
logger.debug("EXIT!")
def add(domain: str, origin: str, command: str, path: str = None, software: str = None):
- logger.debug(f"domain='{domain}',origin='{origin}',command='{command}',path='{path}',software='{software}' - CALLED!")
- if not isinstance(domain, str):
- raise ValueError(f"Parameter domain[]='{type(domain)}' is not 'str'")
- elif domain == "":
- raise ValueError("Parameter 'domain' is empty")
- elif domain.lower() != domain:
- raise ValueError(f"Parameter domain='{domain}' must be all lower-case")
- elif not validators.domain(domain.split("/")[0]):
- raise ValueError(f"domain='{domain}' is not a valid domain")
- elif domain.endswith(".arpa"):
- raise ValueError(f"domain='{domain}' is a domain for reversed IP addresses, please don't crawl them!")
- elif domain.endswith(".tld"):
- raise ValueError(f"domain='{domain}' is a fake domain, please don't crawl them!")
- elif not isinstance(origin, str) and origin is not None:
+ logger.debug("domain='%s',origin='%s',command='%s',path='%s',software='%s' - CALLED!", domain, origin, command, path, software)
+ domain_helper.raise_on(domain)
+
+ if not isinstance(origin, str) and origin is not None:
raise ValueError(f"origin[]='{type(origin)}' is not 'str'")
elif origin == "":
raise ValueError("Parameter 'origin' is empty")
raise ValueError(f"command[]='{type(command)}' is not 'str'")
elif command == "":
raise ValueError("Parameter 'command' is empty")
- elif not validators.domain(domain.split("/")[0]):
- raise ValueError(f"Bad domain name='{domain}'")
elif not isinstance(path, str) and path is not None:
raise ValueError(f"path[]='{type(path)}' is not 'str'")
elif path == "":
raise ValueError(f"software[]='{type(software)}' is not 'str'")
elif software == "":
raise ValueError("Parameter 'software' is empty")
- elif domain.endswith(".arpa"):
- raise ValueError(f"Please don't crawl .arpa domains: domain='{domain}'")
elif origin is not None and not validators.domain(origin.split("/")[0]):
raise ValueError(f"Bad origin name='{origin}'")
elif blacklist.is_blacklisted(domain):
raise Exception(f"domain='{domain}' is blacklisted, but method invoked")
- elif domain.find("/profile/") > 0 or domain.find("/users/") > 0 or (software == "lemmy" and domain.find("/c/") > 0):
+ elif domain.find("/profile/") > 0 or domain.find("/users/") > 0 or (is_registered(domain.split("/")[0]) and domain.find("/c/") > 0):
raise Exception(f"domain='{domain}' is a single user")
- elif domain.endswith(".tld"):
- raise ValueError(f"domain='{domain}' is a fake domain, please don't crawl them!")
+ elif domain.find("/tag/") > 0:
+ raise Exception(f"domain='{domain}' is a tag")
if software is None:
try:
- logger.debug("domain,origin,command,path:", domain, origin, command, path)
+ logger.debug("domain='%s',origin='%s',command='%s',path='%s'", domain, origin, command, path)
software = federation.determine_software(domain, path)
except network.exceptions as exception:
logger.warning("Exception '%s' during determining software type, domain='%s'", type(exception), domain)
logger.warning("domain='%s' already registered after cutting off user part. - EXIT!", domain)
return
- logger.info("Adding instance domain='%s' (origin='%s',software='%s')", domain, origin, software)
+ logger.info("Adding instance domain='%s',origin='%s',software='%s',command='%s'", domain, origin, software, command)
database.cursor.execute(
"INSERT INTO instances (domain, origin, command, hash, software, first_seen) VALUES (?, ?, ?, ?, ?, ?)",
(
),
)
- logger.debug(f"Marking domain='{domain}' as registered.")
+ logger.debug("Marking domain='%s' as registered.", domain)
cache.set_sub_key("is_registered", domain, True)
+ logger.debug("Checking if domain='%s' has pending updates ...", domain)
if has_pending(domain):
- logger.debug(f"domain='{domain}' has pending nodeinfo being updated ...")
+ logger.debug("Flushing updates for domain='%s' ...", domain)
update_data(domain)
logger.debug("EXIT!")
def set_last_nodeinfo(domain: str):
- logger.debug("domain(%d)='%s' - CALLED!", len(domain), domain)
- if not isinstance(domain, str):
- raise ValueError(f"Parameter domain[]='{type(domain)}' is not 'str'")
- elif domain == "":
- raise ValueError("Parameter 'domain' is empty")
- elif domain.lower() != domain:
- raise ValueError(f"Parameter domain='{domain}' must be all lower-case")
- elif not validators.domain(domain.split("/")[0]):
- raise ValueError(f"domain='{domain}' is not a valid domain")
- elif domain.endswith(".arpa"):
- raise ValueError(f"domain='{domain}' is a domain for reversed IP addresses, please don't crawl them!")
- elif domain.endswith(".tld"):
- raise ValueError(f"domain='{domain}' is a fake domain, please don't crawl them!")
+ logger.debug("domain='%s' - CALLED!", domain)
+ domain_helper.raise_on(domain)
- logger.debug("Updating last_nodeinfo for domain:", domain)
+ logger.debug("Updating last_nodeinfo for domain='%s'", domain)
_set_data("last_nodeinfo", domain, time.time())
- # Running pending updated
- logger.debug(f"Invoking update_data({domain}) ...")
- update_data(domain)
-
logger.debug("EXIT!")
def set_last_error(domain: str, error: dict):
- logger.debug("domain,error[]:", domain, type(error))
- if not isinstance(domain, str):
- raise ValueError(f"Parameter domain[]='{type(domain)}' is not 'str'")
- elif domain == "":
- raise ValueError("Parameter 'domain' is empty")
- elif domain.lower() != domain:
- raise ValueError(f"Parameter domain='{domain}' must be all lower-case")
- elif not validators.domain(domain.split("/")[0]):
- raise ValueError(f"domain='{domain}' is not a valid domain")
- elif domain.endswith(".arpa"):
- raise ValueError(f"domain='{domain}' is a domain for reversed IP addresses, please don't crawl them!")
- elif domain.endswith(".tld"):
- raise ValueError(f"domain='{domain}' is a fake domain, please don't crawl them!")
+ logger.debug("domain='%s',error[]='%s' - CALLED!", domain, type(error))
+ domain_helper.raise_on(domain)
- logger.debug("BEFORE error[]:", type(error))
+ logger.debug("error[]='%s' - BEFORE!", type(error))
if isinstance(error, (BaseException, json.decoder.JSONDecodeError)):
error = f"error[{type(error)}]='{str(error)}'"
- logger.debug("AFTER error[]:", type(error))
+ logger.debug("error[]='%s' - AFTER!", type(error))
if isinstance(error, str):
- logger.debug(f"Setting last_error_details='{error}'")
+ logger.debug("Setting last_error_details='%s' (str)", error)
_set_data("last_status_code" , domain, 999)
_set_data("last_error_details", domain, error if error != "" else None)
elif isinstance(error, requests.models.Response):
- logger.debug(f"Setting last_error_details='{error.reason}'")
+ logger.debug("Setting last_error_details='%s' (Response)", error.reason)
_set_data("last_status_code" , domain, error.status_code)
_set_data("last_error_details", domain, error.reason if error.reason != "" else None)
elif not isinstance(error, dict):
raise KeyError(f"Cannot handle keys in error[{type(error)}]='{error}'")
elif "status_code" in error and "error_message" in error:
- logger.debug(f"Setting last_error_details='{error['error_message']}'")
+ logger.debug("Setting last_error_details='%s' (error_message)", error['error_message'])
_set_data("last_status_code" , domain, error["status_code"])
_set_data("last_error_details", domain, error["error_message"] if error["error_message"] != "" else None)
elif "json" in error and "error" in error["json"]:
+ logger.debug("Setting last_error_details='%s' (json,error)", error["json"]["error"])
_set_data("last_status_code" , domain, error["status_code"])
_set_data("last_error_details", domain, error["json"]["error"] if error["json"]["error"] != "" else None)
- logger.debug(f"Invoking error_log.add(domain='{domain}',error[]='{type(error)}'")
+ logger.debug("Invoking error_log.add(domain='%s',error[]='%s'", domain, type(error))
error_log.add(domain, error)
logger.debug("EXIT!")
+def set_success(domain: str):
+ logger.debug("domain='%s' - CALLED!", domain)
+ domain_helper.raise_on(domain)
+
+ # Set both to success
+ _set_data("last_status_code" , domain, 200)
+ _set_data("last_error_details", domain, None)
+
+ logger.debug("EXIT!")
+
def is_registered(domain: str) -> bool:
- logger.debug("domain(%d)='%s' - CALLED!", len(domain), domain)
- if not isinstance(domain, str):
- raise ValueError(f"Parameter domain[]='{type(domain)}' is not 'str'")
- elif domain == "":
- raise ValueError("Parameter 'domain' is empty")
- elif domain.lower() != domain:
- raise ValueError(f"Parameter domain='{domain}' must be all lower-case")
- elif not validators.domain(domain.split("/")[0]):
- raise ValueError(f"domain='{domain}' is not a valid domain")
- elif domain.endswith(".arpa"):
- raise ValueError(f"domain='{domain}' is a domain for reversed IP addresses, please don't crawl them!")
- elif domain.endswith(".tld"):
- raise ValueError(f"domain='{domain}' is a fake domain, please don't crawl them!")
+ logger.debug("domain='%s' - CALLED!", domain)
+ domain_helper.raise_on(domain)
- logger.debug("domain(%d)='%s' - CALLED!", len(domain), domain)
+ logger.debug("domain='%s' - CALLED!", domain)
if not cache.key_exists("is_registered"):
logger.debug("Cache for 'is_registered' not initialized, fetching all rows ...")
database.cursor.execute("SELECT domain FROM instances")
# Is cache found?
registered = cache.sub_key_exists("is_registered", domain)
- logger.debug(f"registered='{registered}' - EXIT!")
+ logger.debug("registered='%s' - EXIT!", registered)
return registered
-def is_recent(domain: str) -> bool:
- logger.debug("domain(%d)='%s' - CALLED!", len(domain), domain)
- if not isinstance(domain, str):
- raise ValueError(f"Parameter domain[]='{type(domain)}' is not 'str'")
- elif domain == "":
- raise ValueError("Parameter 'domain' is empty")
- elif domain.lower() != domain:
- raise ValueError(f"Parameter domain='{domain}' must be all lower-case")
- elif not validators.domain(domain.split("/")[0]):
- raise ValueError(f"domain='{domain}' is not a valid domain")
- elif domain.endswith(".arpa"):
- raise ValueError(f"domain='{domain}' is a domain for reversed IP addresses, please don't crawl them!")
- elif domain.endswith(".tld"):
- raise ValueError(f"domain='{domain}' is a fake domain, please don't crawl them!")
+def is_recent(domain: str, column: str = "last_instance_fetch") -> bool:
+ logger.debug("domain='%s',column='%s' - CALLED!", domain)
+ domain_helper.raise_on(domain)
+
+ if not isinstance(column, str):
+ raise ValueError(f"Parameter column[]='{type(column)}' is not 'str'")
+ elif column not in ["last_instance_fetch", "last_blocked"]:
+ raise ValueError(f"Parameter column='{column}' is not expected")
elif not is_registered(domain):
- logger.debug(f"domain='{domain}' is not registered, returning False - EXIT!")
+ logger.debug("domain='%s' is not registered, returning False - EXIT!", domain)
return False
# Query database
- database.cursor.execute("SELECT last_instance_fetch FROM instances WHERE domain = ? LIMIT 1", [domain])
+ database.cursor.execute(f"SELECT {column} FROM instances WHERE domain = ? LIMIT 1", [domain])
# Fetch row
fetched = database.cursor.fetchone()[0]
- logger.debug(f"fetched[{type(fetched)}]='{fetched}'")
+ logger.debug("fetched[%s]='%s'", type(fetched), fetched)
recently = isinstance(fetched, float) and time.time() - fetched <= config.get("recheck_instance")
- logger.debug(f"recently='{recently}' - EXIT!")
+ logger.debug("recently='%s' - EXIT!", recently)
return recently
-def deobscure(char: str, domain: str, blocked_hash: str = None) -> tuple:
- logger.debug(f"char='{char}',domain='{domain}',blocked_hash='{blocked_hash}' - CALLED!")
- if not isinstance(char, str):
- raise ValueError(f"Parameter char[]='{type(char)}' is not 'str'")
- elif char == "":
- raise ValueError("Parameter 'char' is empty")
- elif not isinstance(domain, str):
+def deobfuscate(char: str, domain: str, blocked_hash: str = None) -> tuple:
+ logger.debug("char='%s',domain='%s',blocked_hash='%s' - CALLED!", char, domain, blocked_hash)
+
+ if not isinstance(domain, str):
raise ValueError(f"Parameter domain[]='{type(domain)}' is not 'str'")
elif domain == "":
raise ValueError("Parameter 'domain' is empty")
elif domain.lower() != domain:
raise ValueError(f"Parameter domain='{domain}' must be all lower-case")
- elif not validators.domain(domain.split("/")[0]):
- raise ValueError(f"domain='{domain}' is not a valid domain")
elif domain.endswith(".arpa"):
raise ValueError(f"domain='{domain}' is a domain for reversed IP addresses, please don't crawl them!")
elif domain.endswith(".tld"):
raise ValueError(f"domain='{domain}' is a fake domain, please don't crawl them!")
+ elif not isinstance(char, str):
+ raise ValueError(f"Parameter char[]='{type(char)}' is not 'str'")
+ elif char == "":
+ raise ValueError("Parameter 'char' is empty")
+ elif not char in domain:
+ raise ValueError(f"char='{char}' not found in domain='{domain}' but function invoked")
elif not isinstance(blocked_hash, str) and blocked_hash is not None:
raise ValueError(f"Parameter blocked_hash[]='{type(blocked_hash)}' is not 'str'")
+ logger.debug("blocked_hash[]='%s'", type(blocked_hash))
if isinstance(blocked_hash, str):
- logger.debug(f"Looking up blocked_hash='{blocked_hash}' ...")
+ logger.debug("Looking up blocked_hash='%s',domain='%s' ...", blocked_hash, domain)
database.cursor.execute(
- "SELECT domain, origin, nodeinfo_url FROM instances WHERE hash = ? LIMIT 1", [blocked_hash]
+ "SELECT domain, origin, nodeinfo_url FROM instances WHERE hash = ? OR domain LIKE ? LIMIT 1", [blocked_hash, domain.replace(char, "_")]
)
row = database.cursor.fetchone()
logger.debug("row[]='%s'", type(row))
if row is None:
- logger.debug(f"blocked_hash='{blocked_hash}' not found, trying domain='{domain}' ...")
- return deobscure(char, domain)
+ logger.debug("blocked_hash='%s' not found, trying domain='%s' ...", blocked_hash, domain)
+ return deobfuscate(char, domain)
else:
- logger.debug(f"Looking up domain='{domain}' ...")
+ logger.debug("Looking up domain='%s' ...", domain)
database.cursor.execute(
"SELECT domain, origin, nodeinfo_url FROM instances WHERE domain LIKE ? ORDER BY rowid LIMIT 1", [domain.replace(char, "_")]
)
row = database.cursor.fetchone()
logger.debug("row[]='%s'", type(row))
- logger.debug(f"row[]='{type(row)}' - EXIT!")
+ logger.debug("row[]='%s' - EXIT!", type(row))
return row
def set_last_blocked(domain: str):
- logger.debug("domain(%d)='%s' - CALLED!", len(domain), domain)
- if not isinstance(domain, str):
- raise ValueError(f"Parameter domain[]='{type(domain)}' is not 'str'")
- elif domain == "":
- raise ValueError("Parameter 'domain' is empty")
- elif domain.lower() != domain:
- raise ValueError(f"Parameter domain='{domain}' must be all lower-case")
- elif not validators.domain(domain.split("/")[0]):
- raise ValueError(f"domain='{domain}' is not a valid domain")
- elif domain.endswith(".arpa"):
- raise ValueError(f"domain='{domain}' is a domain for reversed IP addresses, please don't crawl them!")
- elif domain.endswith(".tld"):
- raise ValueError(f"domain='{domain}' is a fake domain, please don't crawl them!")
+ logger.debug("domain='%s' - CALLED!", domain)
+ domain_helper.raise_on(domain)
# Set timestamp
_set_data("last_blocked", domain, time.time())
logger.debug("EXIT!")
def set_last_instance_fetch(domain: str):
- logger.debug("domain(%d)='%s' - CALLED!", len(domain), domain)
- if not isinstance(domain, str):
- raise ValueError(f"Parameter domain[]='{type(domain)}' is not 'str'")
- elif domain == "":
- raise ValueError("Parameter 'domain' is empty")
- elif domain.lower() != domain:
- raise ValueError(f"Parameter domain='{domain}' must be all lower-case")
- elif not validators.domain(domain.split("/")[0]):
- raise ValueError(f"domain='{domain}' is not a valid domain")
- elif domain.endswith(".arpa"):
- raise ValueError(f"domain='{domain}' is a domain for reversed IP addresses, please don't crawl them!")
- elif domain.endswith(".tld"):
- raise ValueError(f"domain='{domain}' is a fake domain, please don't crawl them!")
+ logger.debug("domain='%s' - CALLED!", domain)
+ domain_helper.raise_on(domain)
# Set timestamp
_set_data("last_instance_fetch", domain, time.time())
logger.debug("EXIT!")
def set_total_peers(domain: str, peers: list):
- logger.debug(f"domain='{domain}',peers()={len(peers)} - CALLED!")
- if not isinstance(domain, str):
- raise ValueError(f"Parameter domain[]='{type(domain)}' is not 'str'")
- elif domain == "":
- raise ValueError("Parameter 'domain' is empty")
- elif domain.lower() != domain:
- raise ValueError(f"Parameter domain='{domain}' must be all lower-case")
- elif not validators.domain(domain.split("/")[0]):
- raise ValueError(f"domain='{domain}' is not a valid domain")
- elif domain.endswith(".arpa"):
- raise ValueError(f"domain='{domain}' is a domain for reversed IP addresses, please don't crawl them!")
- elif domain.endswith(".tld"):
- raise ValueError(f"domain='{domain}' is a fake domain, please don't crawl them!")
- elif not isinstance(peers, list):
- raise ValueError(f"Parameter peers[]='{type(peers)}' is not 'list'")
+ logger.debug("domain='%s',peers()=%d - CALLED!", domain, len(peers))
+ domain_helper.raise_on(domain)
+
+ if not isinstance(peers, list):
+ raise ValueError(f"Parameter peers[]='{type(peers)}' is not 'list': '%s'")
# Set timestamp
_set_data("total_peers", domain, len(peers))
logger.debug("EXIT!")
def set_nodeinfo_url(domain: str, url: str):
- logger.debug(f"domain='{domain}',url='{url}' - CALLED!")
- if not isinstance(domain, str):
- raise ValueError(f"Parameter domain[]='{type(domain)}' is not 'str'")
- elif domain == "":
- raise ValueError("Parameter 'domain' is empty")
- elif domain.lower() != domain:
- raise ValueError(f"Parameter domain='{domain}' must be all lower-case")
- elif not validators.domain(domain.split("/")[0]):
- raise ValueError(f"domain='{domain}' is not a valid domain")
- elif domain.endswith(".arpa"):
- raise ValueError(f"domain='{domain}' is a domain for reversed IP addresses, please don't crawl them!")
- elif domain.endswith(".tld"):
- raise ValueError(f"domain='{domain}' is a fake domain, please don't crawl them!")
- elif not isinstance(url, str):
+ logger.debug("domain='%s',url='%s' - CALLED!", domain, url)
+ domain_helper.raise_on(domain)
+
+ if not isinstance(url, str):
raise ValueError("Parameter url[]='{type(url)}' is not 'list'")
elif url == "":
raise ValueError("Parameter 'url' is empty")
logger.debug("EXIT!")
def set_detection_mode(domain: str, mode: str):
- logger.debug(f"domain='{domain}',mode='{mode}' - CALLED!")
- if not isinstance(domain, str):
- raise ValueError(f"Parameter domain[]='{type(domain)}' is not 'str'")
- elif domain == "":
- raise ValueError("Parameter 'domain' is empty")
- elif domain.lower() != domain:
- raise ValueError(f"Parameter domain='{domain}' must be all lower-case")
- elif not validators.domain(domain.split("/")[0]):
- raise ValueError(f"domain='{domain}' is not a valid domain")
- elif domain.endswith(".arpa"):
- raise ValueError(f"domain='{domain}' is a domain for reversed IP addresses, please don't crawl them!")
- elif domain.endswith(".tld"):
- raise ValueError(f"domain='{domain}' is a fake domain, please don't crawl them!")
- elif not isinstance(mode, str):
+ logger.debug("domain='%s',mode='%s' - CALLED!", domain, mode)
+ domain_helper.raise_on(domain)
+
+ if not isinstance(mode, str):
raise ValueError("Parameter mode[]='{type(mode)}' is not 'list'")
elif mode == "":
raise ValueError("Parameter 'mode' is empty")
# Set timestamp
_set_data("detection_mode", domain, mode)
logger.debug("EXIT!")
+
+def set_has_obfuscation(domain: str, status: bool):
+ logger.debug("domain(%d)='%s',status='%s' - CALLED!", len(domain), domain, status)
+ domain_helper.raise_on(domain)
+
+ if not isinstance(status, bool):
+ raise ValueError(f"Parameter status[]='{type(status)}' is not 'bool'")
+
+ # Set timestamp
+ _set_data("has_obfuscation", domain, status)
+ logger.debug("EXIT!")
projects / fba.git / blobdiff