ip user info
This commit is contained in:
75
ip_analyzer_app/services/analysis.py
Normal file
75
ip_analyzer_app/services/analysis.py
Normal file
@@ -0,0 +1,75 @@
|
||||
from collections import Counter
|
||||
|
||||
from .lookups import cymru_lookup, whois_lookup
|
||||
from .parsing import parse_whois
|
||||
|
||||
|
||||
def analyze_ip(ip: str, cymru_data: dict[str, dict[str, str]] | None = None) -> dict[str, str]:
|
||||
info = {
|
||||
'ip': ip,
|
||||
'asn': 'Unknown',
|
||||
'owner': 'Unknown',
|
||||
'user': 'Unknown',
|
||||
'country': 'Unknown',
|
||||
'network': 'Unknown',
|
||||
}
|
||||
|
||||
if cymru_data and ip in cymru_data:
|
||||
data = cymru_data[ip]
|
||||
info['asn'] = data.get('asn', 'Unknown')
|
||||
info['owner'] = data.get('owner', 'Unknown')
|
||||
info['country'] = data.get('country', 'Unknown')
|
||||
info['network'] = data.get('prefix', 'Unknown')
|
||||
|
||||
whois_output = whois_lookup(ip)
|
||||
if whois_output:
|
||||
parsed = parse_whois(whois_output)
|
||||
if info['asn'] == 'Unknown' and parsed['asn'] != 'Unknown':
|
||||
info['asn'] = parsed['asn']
|
||||
if parsed['country'] != 'Unknown':
|
||||
info['country'] = parsed['country']
|
||||
if parsed['cidr'] != 'Unknown':
|
||||
info['network'] = parsed['cidr']
|
||||
info['user'] = parsed['user']
|
||||
if info['owner'] == 'Unknown':
|
||||
info['owner'] = parsed['org'] if parsed['org'] != 'Unknown' else parsed['netname']
|
||||
|
||||
return info
|
||||
|
||||
|
||||
def analyze_ips(ips: list[str]) -> tuple[list[dict[str, str]], dict[str, dict[str, int]]]:
|
||||
print(f'Analyzing {len(ips)} IPs via Team Cymru...')
|
||||
cymru_data = cymru_lookup(ips)
|
||||
|
||||
results = [analyze_ip(ip, cymru_data) for ip in ips]
|
||||
stats = {
|
||||
'total': len(results),
|
||||
'countries': dict(Counter(r['country'] for r in results)),
|
||||
'asns': dict(Counter(r['asn'] for r in results)),
|
||||
'owners': dict(Counter(r['owner'] for r in results)),
|
||||
'users': dict(Counter(r['user'] for r in results)),
|
||||
}
|
||||
|
||||
print(f"Analysis complete: {len(results)} IPs, {len(stats['countries'])} countries")
|
||||
return results, stats
|
||||
|
||||
|
||||
def apply_filters(results: list[dict[str, str]], filters: dict) -> list[dict[str, str]]:
|
||||
countries = set(filters.get('countries', []))
|
||||
asns = set(filters.get('asns', []))
|
||||
owners = set(filters.get('owners', []))
|
||||
users = set(filters.get('users', []))
|
||||
|
||||
if not (countries or asns or owners or users):
|
||||
return results
|
||||
|
||||
filtered: list[dict[str, str]] = []
|
||||
for item in results:
|
||||
if (
|
||||
(not countries or item['country'] in countries)
|
||||
and (not asns or item['asn'] in asns)
|
||||
and (not owners or item['owner'] in owners)
|
||||
and (not users or item.get('user', 'Unknown') in users)
|
||||
):
|
||||
filtered.append(item)
|
||||
return filtered
|
||||
Reference in New Issue
Block a user