-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Removes unused utility code and moves around single use code
- Fixes docstrings without linebreak when multiline - Removes utilities and moves them to scripts using them - Adds a test to iprange plugin - Moves domain utils to certcheck plugin
- Loading branch information
Showing
16 changed files
with
249 additions
and
288 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,70 @@ | ||
from collections import deque | ||
from typing import Iterable, TypeVar, Hashable | ||
|
||
T = TypeVar('T') | ||
R = TypeVar('R') | ||
H = TypeVar('H', bound=Hashable) | ||
|
||
|
||
def lagging_filter(lag: int): | ||
""" | ||
Filters our values from an iterator that were recently seen | ||
The memory works using a first-in-first-out (FIFO) deque | ||
""" | ||
values = set() | ||
latest = deque() | ||
|
||
if lag <= 0: | ||
raise ValueError('Invalid lag value') | ||
|
||
def filter_function(item): | ||
if item not in values: | ||
if len(values) >= lag: | ||
values.remove(latest.popleft()) | ||
latest.append(item) | ||
values.add(item) | ||
return True | ||
else: | ||
return False | ||
|
||
return filter_function | ||
|
||
|
||
def filter_locally_unique(iterable: Iterable[H], *, lag: int) -> Iterable[H]: | ||
""" | ||
Filters iterable to produce locally unique values | ||
This could be useful for removing duplicate values from a very long, | ||
but relatively sorted iterable. | ||
Requires the iterable values to be hashable. | ||
""" | ||
return filter(lagging_filter(lag), iterable) | ||
|
||
|
||
def spread_domain(domain: str) -> Iterable[str]: | ||
"""Spreads a domain into all valid super domains | ||
""" | ||
parts = domain.strip().split('.') | ||
if parts[0] == '*' or parts[0] == '': | ||
parts = parts[1:] | ||
if len(parts) > 2: | ||
partial_domain = '.'.join(parts[-2:]) | ||
yield partial_domain | ||
for i in range(-3, -len(parts) - 1, -1): | ||
partial_domain = f'{parts[i]}.{partial_domain}' | ||
yield partial_domain | ||
elif len(parts) == 2: | ||
yield '.'.join(parts) | ||
|
||
|
||
def process_domains(domains: Iterable[str], *, lag: int = 100) -> Iterable[str]: | ||
"""Best effort de-duplicates and removes star domains from the input and creates all valid super domains | ||
""" | ||
return filter_locally_unique((value for domain in domains for value in spread_domain(domain)), lag=lag) | ||
|
||
|
||
__all__ = [ | ||
'process_domains', | ||
] |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,69 @@ | ||
import pytest | ||
|
||
from dnsmule_plugins.certcheck.domains import lagging_filter, filter_locally_unique, process_domains, spread_domain | ||
|
||
|
||
@pytest.mark.parametrize('i,lag,result', [ | ||
( | ||
['a', 'a', 'a', 'b', 'b'], | ||
1, | ||
['a', 'b'], | ||
), | ||
( | ||
['a', 'b', 'c', 'a', 'b', 'c'], | ||
3, | ||
['a', 'b', 'c'], | ||
), | ||
( | ||
['a', 'b', 'c', 'd', 'a', 'b', 'c', 'd'], | ||
3, | ||
['a', 'b', 'c', 'd', 'a', 'b', 'c', 'd'], | ||
), | ||
( | ||
['a', 'b', 'c', 'd', 'a', 'b', 'c', 'd'], | ||
2, | ||
['a', 'b', 'c', 'd', 'a', 'b', 'c', 'd'], | ||
), | ||
( | ||
['a', 'b', 'c', 'd', 'a', 'a', 'b', 'b'], | ||
2, | ||
['a', 'b', 'c', 'd', 'a', 'b'], | ||
), | ||
]) | ||
def test_lagging_filter(i, lag, result): | ||
assert [*filter_locally_unique(i, lag=lag)] == result, 'Did not filter' | ||
|
||
|
||
@pytest.mark.parametrize('lag', [ | ||
0, | ||
-1, | ||
]) | ||
def test_lagging_filter_value_error(lag): | ||
with pytest.raises(ValueError): | ||
lagging_filter(lag) | ||
|
||
|
||
@pytest.mark.parametrize('value,result', [ | ||
( | ||
['a.b', 'b.c', 'd.e', 'a.e'], | ||
['a.b', 'b.c', 'd.e', 'a.e'], | ||
), | ||
( | ||
['a.b', '*.a.b', 'a.b', 'b.c.e'], | ||
['a.b', 'b.c.e', 'c.e'], | ||
) | ||
]) | ||
def test_process_domains(value, result): | ||
assert set(process_domains(value)) == set(result), 'Unexpected result' | ||
|
||
|
||
def test_spread_domain_skips_tld(): | ||
assert [*spread_domain('.fi')] == [], 'Failed to skip TLD' | ||
|
||
|
||
def test_spread_domain_short(): | ||
assert [*spread_domain('a.fi')] == ['a.fi'], 'Failed to short' | ||
|
||
|
||
def test_spread_domain(): | ||
assert [*spread_domain('a.b.c.fi')] == ['c.fi', 'b.c.fi', 'a.b.c.fi'], 'Failed to produce all domains' |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -1,4 +1,2 @@ | ||
from .baseclasses import * | ||
from .misc import * | ||
from .domains import * | ||
from .iterables import * |
This file was deleted.
Oops, something went wrong.
This file was deleted.
Oops, something went wrong.
Oops, something went wrong.