]> git.mxchange.org Git - fba.git/blob - fba/models/blocks.py
Continued:
[fba.git] / fba / models / blocks.py
1 # Fedi API Block - An aggregator for fetching blocking data from fediverse nodes
2 # Copyright (C) 2023 Free Software Foundation
3 #
4 # This program is free software: you can redistribute it and/or modify
5 # it under the terms of the GNU Affero General Public License as published
6 # by the Free Software Foundation, either version 3 of the License, or
7 # (at your option) any later version.
8 #
9 # This program is distributed in the hope that it will be useful,
10 # but WITHOUT ANY WARRANTY; without even the implied warranty of
11 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
12 # GNU Affero General Public License for more details.
13 #
14 # You should have received a copy of the GNU Affero General Public License
15 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
16
17 import time
18 import validators
19
20 from fba import blacklist
21 from fba import fba
22 from fba.helpers import tidyup
23
24 def update_reason(reason: str, blocker: str, blocked: str, block_level: str):
25     # DEBUG: print(f"DEBUG: reason='{reason}',blocker={blocker},blocked={blocked},block_level={block_level} - CALLED!")
26     if not isinstance(reason, str) and reason is not None:
27         raise ValueError(f"Parameter reason[]='{type(reason)}' is not 'str'")
28     elif not isinstance(blocker, str):
29         raise ValueError(f"Parameter blocker[]='{type(blocker)}' is not 'str'")
30     elif blocker == "":
31         raise ValueError("Parameter 'blocker' is empty")
32     elif not isinstance(blocked, str):
33         raise ValueError(f"Parameter blocked[]='{type(blocked)}' is not 'str'")
34     elif blocked == "":
35         raise ValueError("Parameter 'blocked' is empty")
36     elif not isinstance(block_level, str):
37         raise ValueError(f"Parameter block_level[]='{type(block_level)}' is not 'str'")
38     elif block_level == "":
39         raise ValueError("Parameter 'block_level' is empty")
40     elif block_level == "accept":
41         raise ValueError("Accepted domains are not wanted here")
42
43     # DEBUG: print("DEBUG: Updating block reason:", reason, blocker, blocked, block_level)
44     fba.cursor.execute(
45         "UPDATE blocks SET reason = ?, last_seen = ? WHERE blocker = ? AND blocked = ? AND block_level = ? AND (reason IS NULL OR reason = '') LIMIT 1",
46         [
47             reason,
48             time.time(),
49             blocker,
50             blocked,
51             block_level
52         ])
53
54     # DEBUG: print("DEBUG: EXIT!")
55
56 def update_last_seen(blocker: str, blocked: str, block_level: str):
57     # DEBUG: print("DEBUG: Updating last_seen for:", blocker, blocked, block_level)
58     if not isinstance(blocker, str):
59         raise ValueError(f"Parameter blocker[]='{type(blocker)}' is not 'str'")
60     elif blocker == "":
61         raise ValueError("Parameter 'blocker' is empty")
62     elif not isinstance(blocked, str):
63         raise ValueError(f"Parameter blocked[]='{type(blocked)}' is not 'str'")
64     elif blocked == "":
65         raise ValueError("Parameter 'blocked' is empty")
66     elif not isinstance(block_level, str):
67         raise ValueError(f"Parameter block_level[]='{type(block_level)}' is not 'str'")
68     elif block_level == "":
69         raise ValueError("Parameter 'block_level' is empty")
70     elif block_level == "accept":
71         raise ValueError("Accepted domains are not wanted here")
72
73     fba.cursor.execute(
74         "UPDATE blocks SET last_seen = ? WHERE blocker = ? AND blocked = ? AND block_level = ? LIMIT 1",
75         [
76             time.time(),
77             blocker,
78             blocked,
79             block_level
80         ])
81
82     # DEBUG: print("DEBUG: EXIT!")
83
84 def is_instance_blocked(blocker: str, blocked: str, block_level: str) -> bool:
85     # DEBUG: print(f"DEBUG: blocker={blocker},blocked={blocked},block_level={block_level} - CALLED!")
86     if not isinstance(blocker, str):
87         raise ValueError(f"Parameter blocker[]='{type(blocker)}' is not of type 'str'")
88     elif blocker == "":
89         raise ValueError("Parameter 'blocker' is empty")
90     elif not isinstance(blocked, str):
91         raise ValueError(f"Parameter blocked[]='{type(blocked)}' is not of type 'str'")
92     elif blocked == "":
93         raise ValueError("Parameter 'blocked' is empty")
94     elif not isinstance(block_level, str):
95         raise ValueError(f"Parameter block_level[]='{type(block_level)}' is not of type 'str'")
96     elif block_level == "":
97         raise ValueError("Parameter 'block_level' is empty")
98     elif block_level == "accept":
99         raise ValueError("Accepted domains are not wanted here")
100
101     fba.cursor.execute(
102         "SELECT * FROM blocks WHERE blocker = ? AND blocked = ? AND block_level = ? LIMIT 1",
103         (
104             blocker,
105             blocked,
106             block_level
107         ),
108     )
109
110     is_blocked = fba.cursor.fetchone() is not None
111
112     # DEBUG: print(f"DEBUG: is_blocked='{is_blocked}' - EXIT!")
113     return is_blocked
114
115 def add_instance(blocker: str, blocked: str, reason: str, block_level: str):
116     # DEBUG: print("DEBUG: blocker,blocked,reason,block_level:", blocker, blocked, reason, block_level)
117     if not isinstance(blocker, str):
118         raise ValueError(f"Parameter blocker[]='{type(blocker)}' is not 'str'")
119     elif blocker == "":
120         raise ValueError("Parameter 'blocker' is empty")
121     elif not validators.domain(blocker.split("/")[0]):
122         raise ValueError(f"Bad blocker='{blocker}'")
123     elif not isinstance(blocked, str):
124         raise ValueError(f"Parameter blocked[]='{type(blocked)}' is not 'str'")
125     elif blocked == "":
126         raise ValueError("Parameter 'blocked' is empty")
127     elif not isinstance(block_level, str):
128         raise ValueError(f"Parameter block_level[]='{type(block_level)}' is not of type 'str'")
129     elif block_level == "":
130         raise ValueError("Parameter 'block_level' is empty")
131     elif not validators.domain(blocked.split("/")[0]):
132         raise ValueError(f"Bad blocked='{blocked}'")
133     elif blacklist.is_blacklisted(blocker):
134         raise Exception(f"blocker='{blocker}' is blacklisted but function invoked")
135     elif blacklist.is_blacklisted(blocked):
136         raise Exception(f"blocked='{blocked}' is blacklisted but function invoked")
137     elif block_level == "accept":
138         raise ValueError("Accepted domains are not wanted here")
139
140     if reason is not None:
141         # Maybe needs cleaning
142         reason = tidyup.reason(reason)
143
144     print(f"INFO: New block: blocker='{blocker}',blocked='{blocked}',reason='{reason}',block_level='{block_level}'")
145
146     fba.cursor.execute(
147         "INSERT INTO blocks (blocker, blocked, reason, block_level, first_seen, last_seen) VALUES (?, ?, ?, ?, ?, ?)",
148         [
149              blocker,
150              blocked,
151              reason,
152              block_level,
153              time.time(),
154              time.time()
155         ])
156
157     # DEBUG: print("DEBUG: EXIT!")