-def update_block_reason(reason: str, blocker: str, blocked: str, block_level: str):
- # DEBUG: print(f"DEBUG: reason='{reason}',blocker={blocker},blocked={blocked},block_level={block_level} - CALLED!")
- if type(reason) != str and reason != None:
- raise ValueError(f"Parameter reason[]='{type(reason)}' is not 'str'")
- elif type(blocker) != str:
- raise ValueError(f"Parameter blocker[]='{type(blocker)}' is not 'str'")
- elif type(blocked) != str:
- raise ValueError(f"Parameter blocked[]='{type(blocked)}' is not 'str'")
- elif type(block_level) != str:
- raise ValueError(f"Parameter block_level[]='{type(block_level)}' is not 'str'")
-
- # DEBUG: print("DEBUG: Updating block reason:", reason, blocker, blocked, block_level)
- try:
- cursor.execute(
- "UPDATE blocks SET reason = ?, last_seen = ? WHERE blocker = ? AND blocked = ? AND block_level = ? AND reason IN ('','unknown') LIMIT 1",
- (
- reason,
- time.time(),
- blocker,
- blocked,
- block_level
- ),
- )
-
- # DEBUG: print(f"DEBUG: cursor.rowcount={cursor.rowcount}")
- if cursor.rowcount == 0:
- # DEBUG: print(f"DEBUG: Did not update any rows: blocker='{blocker}',blocked='{blocked}',block_level='{block_level}',reason='{reason}' - EXIT!")
- return
-
- except BaseException as e:
- print(f"ERROR: failed SQL query: reason='{reason}',blocker='{blocker}',blocked='{blocked}',block_level='{block_level}',exception[{type(e)}]:'{str(e)}'")
- sys.exit(255)
-
- # DEBUG: print("DEBUG: EXIT!")
-
-def update_last_seen(blocker: str, blocked: str, block_level: str):
- # DEBUG: print("DEBUG: Updating last_seen for:", blocker, blocked, block_level)
- try:
- cursor.execute(
- "UPDATE blocks SET last_seen = ? WHERE blocker = ? AND blocked = ? AND block_level = ? LIMIT 1",
- (
- time.time(),
- blocker,
- blocked,
- block_level
- )
- )
-
- # DEBUG: print(f"DEBUG: cursor.rowcount={cursor.rowcount}")
- if cursor.rowcount == 0:
- # DEBUG: print(f"DEBUG: Did not update any rows: blocker='{blocker}',blocked='{blocked}',block_level='{block_level}' - EXIT!")
- return
-
- except BaseException as e:
- print(f"ERROR: failed SQL query: blocker='{blocker}',blocked='{blocked}',block_level='{block_level}',exception[{type(e)}]:'{str(e)}'")
- sys.exit(255)
-
- # DEBUG: print("DEBUG: EXIT!")
-
-def is_instance_blocked(blocker: str, blocked: str, block_level: str) -> bool:
- # DEBUG: print(f"DEBUG: blocker={blocker},blocked={blocked},block_level={block_level} - CALLED!")
- if type(blocker) != str:
- raise ValueError(f"Parameter blocker[]={type(blocker)} is not of type 'str'")
- elif blocker == "":
- raise ValueError("Parameter 'blocker' cannot be empty")
- elif type(blocked) != str:
- raise ValueError(f"Parameter blocked[]={type(blocked)} is not of type 'str'")
- elif blocked == "":
- raise ValueError("Parameter 'blocked' cannot be empty")
- elif type(block_level) != str:
- raise ValueError(f"Parameter block_level[]={type(block_level)} is not of type 'str'")
- elif block_level == "":
- raise ValueError("Parameter 'block_level' cannot be empty")
-
- cursor.execute(
- "SELECT * FROM blocks WHERE blocker = ? AND blocked = ? AND block_level = ? LIMIT 1",
- (
- blocker,
- blocked,
- block_level
- ),
- )
-
- is_blocked = cursor.fetchone() != None
-
- # DEBUG: print(f"DEBUG: is_blocked='{is_blocked}' - EXIT!")
- return is_blocked
-
-def block_instance(blocker: str, blocked: str, reason: str, block_level: str):
- # DEBUG: print("DEBUG: blocker,blocked,reason,block_level:", blocker, blocked, reason, block_level)
- if type(blocker) != str:
- raise ValueError(f"Parameter blocker[]={type(blocker)} is not 'str'")
- elif blocker == "":
- raise ValueError(f"Parameter 'blocker' cannot be empty")
- elif not validators.domain(blocker.split("/")[0]):
- raise ValueError(f"Bad blocker='{blocker}'")
- elif type(blocked) != str:
- raise ValueError(f"Parameter blocked[]={type(blocked)} is not 'str'")
- elif blocked == "":
- raise ValueError(f"Parameter 'blocked' cannot be empty")
- elif not validators.domain(blocked.split("/")[0]):
- raise ValueError(f"Bad blocked='{blocked}'")
- elif is_blacklisted(blocker):
- raise Exception(f"blocker='{blocker}' is blacklisted but function invoked")
- elif is_blacklisted(blocked):
- raise Exception(f"blocked='{blocked}' is blacklisted but function invoked")
-
- if reason != None:
- # Maybe needs cleaning
- reason = tidyup_reason(reason)
-
- print(f"INFO: New block: blocker='{blocker}',blocked='{blocked}', reason='{reason}', block_level='{block_level}'")
- try:
- cursor.execute(
- "INSERT INTO blocks (blocker, blocked, reason, block_level, first_seen, last_seen) VALUES(?, ?, ?, ?, ?, ?)",
- (
- blocker,
- blocked,
- reason,
- block_level,
- time.time(),
- time.time()
- ),
- )
- except BaseException as e:
- print(f"ERROR: failed SQL query: blocker='{blocker}',blocked='{blocked}',reason='{reason}',block_level='{block_level}',exception[{type(e)}]:'{str(e)}'")
- sys.exit(255)
-
- # DEBUG: print("DEBUG: EXIT!")
-
-def is_instance_registered(domain: str) -> bool:
- # DEBUG: print(f"DEBUG: domain={domain} - CALLED!")