1
1
#!/usr/bin/env python3
2
2
3
- from ipaddress import IPv4Address , IPv6Address , AddressValueError , IPv4Interface , IPv6Interface
3
+ from ipaddress import IPv4Interface , IPv6Interface
4
4
from re import sub as regex_replace
5
5
from threading import Lock
6
6
from json import dumps as json_dumps
13
13
from flask import Flask , request , Response , json , redirect
14
14
from waitress import serve
15
15
import maxminddb
16
- from oxl_utils .valid .net import valid_ip4 , valid_public_ip , valid_asn
16
+ from oxl_utils .valid .net import valid_ip4 , valid_public_ip , valid_asn , get_ipv
17
17
18
18
app = Flask ('risk-db' )
19
19
BASE_DIR = Path ('/var/local/lib/risk-db' )
@@ -46,13 +46,6 @@ def _response_json(code: int, data: dict) -> Response:
46
46
)
47
47
48
48
49
- def _get_ipv (ip : str ) -> int :
50
- if valid_ip4 (ip ):
51
- return 4
52
-
53
- return 6
54
-
55
-
56
49
def _get_src_ip () -> str :
57
50
if valid_public_ip (request .remote_addr ):
58
51
return request .remote_addr
@@ -114,7 +107,7 @@ def check(ip) -> Response:
114
107
return _response_json (code = 400 , data = {'msg' : 'Invalid IP provided' })
115
108
116
109
try :
117
- with maxminddb .open_database (RISKY_DB_FILE [_get_ipv (ip )]) as m :
110
+ with maxminddb .open_database (RISKY_DB_FILE [get_ipv (ip )]) as m :
118
111
r = m .get (ip )
119
112
if r is None :
120
113
return _response_json (code = 404 , data = {'msg' : 'Provided IP not reported' })
@@ -136,7 +129,7 @@ def check_net(ip) -> Response:
136
129
if not valid_public_ip (ip ):
137
130
return _response_json (code = 400 , data = {'msg' : 'Invalid IP provided' })
138
131
139
- ipv = _get_ipv (ip )
132
+ ipv = get_ipv (ip )
140
133
141
134
if ipv == 4 :
142
135
net = IPv4Interface (f"{ ip } /{ NET_SIZE [ipv ]} " ).network .network_address .compressed
0 commit comments