<commit_before>"""
Common Output Format for passive DNS library.

Copyright 2021: Farsight Security (https://www.farsightsecurity.com/)

Author: <NAME> <<EMAIL>>

Released under the Apache 2.0 license.
See: https://www.apache.org/licenses/LICENSE-2.0.txt

"""

import ipaddress
import sys
import ndjson


def is_valid_ip(ip: str) -> bool:
    """Check if an IP address given as string would be convertible to
    an ipaddress object (and thus if it is a valid IP).

    Returns
    --------
    True on success, False on validation failure.
    """

    try:
        ipaddress.ip_address(ip)
    except Exception as ex:
        print("is_valid_ip(%s) returned False. Reason: %s" % (ip, str(ex)), file=sys.stderr)
        return False
    return True


def is_cof_valid_strict(d: dict) -> bool:
    """Check the COF - do the full JSON schema validation.

    Returns
    --------
    True on success, False on validation failure.
    """
    return True     # FIXME


def is_cof_valid_simple(d: dict) -> bool:
    """Check MANDATORY fields according to COF - simple check, do not do the full JSON schema validation.

    Returns
    --------
    True on success, False on validation failure.
    """

    if "rrname" not in d:
        print("Missing MANDATORY field 'rrname'", file=sys.stderr)
        return False
    if not isinstance(d['rrname'], str):
        print("Type error: 'rrname' is not a JSON string", file=sys.stderr)
        return False
    if "rrtype" not in d:
        print("Missing MANDATORY field 'rrtype'", file=sys.stderr)
        return False
    if not isinstance(d['rrtype'], str):
        print("Type error: 'rrtype' is not a JSON string", file=sys.stderr)
        return False
    if "rdata" not in d:
        print("Missing MANDATORY field 'rdata'", file=sys.stderr)
        return False
    if "rdata" not in d:
        print("Missing MANDATORY field 'rdata'", file=sys.stderr)
        return False
    if not isinstance(d['rdata'], str) and not isinstance(d['rdata'], list):
        print("'rdata' is not a list and not a string.", file=sys.stderr)
        return False
    if not ("time_first" in d and "time_last" in d) or ("zone_time_first" in d and "zone_time_last" in d):
        print("We are missing EITHER ('first_seen' and 'last_seen') OR ('zone_time_first' and zone_time_last') fields", file=sys.stderr)
        return False
    # currently we don't check the OPTIONAL fields. Sorry... to be done later.
    return True


def validate_cof(d: dict, strict=True) -> bool:
    """Validate an input passive DNS COF (given as dict).
    strict might be set to False in order to loosen the checking.
    With strict==True, a full JSON Schema validation will happen.


    Returns
    --------
    True on success, False on validation failure.
    """
    if not strict:
        return is_cof_valid_simple(d)
    else:
        return is_cof_valid_strict(d)


if __name__ == "__main__":
    # simple, poor man's unit tests.

    print(80 * "=", file=sys.stderr)
    print("Unit Tests:", file=sys.stderr)
    assert not is_valid_ip("a.2.3.4")<commit_msg>Add a function to validate dnsdbflex output<commit_after>"""
Common Output Format for passive DNS library.

Copyright 2021: Farsight Security (https://www.farsightsecurity.com/)

Author: <NAME> <<EMAIL>>

Released under the Apache 2.0 license.
See: https://www.apache.org/licenses/LICENSE-2.0.txt

"""

import ipaddress
import sys
import ndjson


def is_valid_ip(ip: str) -> bool:
    """Check if an IP address given as string would be convertible to
    an ipaddress object (and thus if it is a valid IP).

    Returns
    --------
    True on success, False on validation failure.
    """

    try:
        ipaddress.ip_address(ip)
    except Exception as ex:
        print("is_valid_ip(%s) returned False. Reason: %s" % (ip, str(ex)), file=sys.stderr)
        return False
    return True


def is_cof_valid_strict(d: dict) -> bool:
    """Check the COF - do the full JSON schema validation.

    Returns
    --------
    True on success, False on validation failure.
    """
    return True     # FIXME


def is_cof_valid_simple(d: dict) -> bool:
    """Check MANDATORY fields according to COF - simple check, do not do the full JSON schema validation.

    Returns
    --------
    True on success, False on validation failure.
    """

    if "rrname" not in d:
        print("Missing MANDATORY field 'rrname'", file=sys.stderr)
        return False
    if not isinstance(d['rrname'], str):
        print("Type error: 'rrname' is not a JSON string", file=sys.stderr)
        return False
    if "rrtype" not in d:
        print("Missing MANDATORY field 'rrtype'", file=sys.stderr)
        return False
    if not isinstance(d['rrtype'], str):
        print("Type error: 'rrtype' is not a JSON string", file=sys.stderr)
        return False
    if "rdata" not in d:
        print("Missing MANDATORY field 'rdata'", file=sys.stderr)
        return False
    if "rdata" not in d:
        print("Missing MANDATORY field 'rdata'", file=sys.stderr)
        return False
    if not isinstance(d['rdata'], str) and not isinstance(d['rdata'], list):
        print("'rdata' is not a list and not a string.", file=sys.stderr)
        return False
    if not ("time_first" in d and "time_last" in d) or ("zone_time_first" in d and "zone_time_last" in d):
        print("We are missing EITHER ('first_seen' and 'last_seen') OR ('zone_time_first' and zone_time_last') fields", file=sys.stderr)
        return False
    # currently we don't check the OPTIONAL fields. Sorry... to be done later.
    return True


def validate_cof(d: dict, strict=True) -> bool:
    """Validate an input passive DNS COF (given as dict).
    strict might be set to False in order to loosen the checking.
    With strict==True, a full JSON Schema validation will happen.


    Returns
    --------
    True on success, False on validation failure.
    """
    if not strict:
        return is_cof_valid_simple(d)
    else:
        return is_cof_valid_strict(d)

def validate_dnsdbflex(d: dict, strict=True) -> bool:
    """
    Validate if dict d is valid dnsdbflex. It should looks like this:
    { "rrtype": <str>, "rrname": <str> }
    """
    if "rrname" not in d:
        print("Missing MANDATORY field 'rrname'", file=sys.stderr)
        return False
    if not isinstance(d['rrname'], str):
        print("Type error: 'rrname' is not a JSON string", file=sys.stderr)
        return False
    if "rrtype" not in d:
        print("Missing MANDATORY field 'rrtype'", file=sys.stderr)
        return False
    if not isinstance(d['rrtype'], str):
        print("Type error: 'rrtype' is not a JSON string", file=sys.stderr)
        return False
    return True


if __name__ == "__main__":
    # simple, poor man's unit tests.

    print(80 * "=", file=sys.stderr)
    print("Unit Tests:", file=sys.stderr)
    assert not is_valid_ip("a.2.3.4")
{'fraction_non_alphanumeric': 0.11605331787887568, 'fraction_numerical': 0.0037670240509997104, 'mean_word_length': 2.7516304347826086, 'pattern_counts': {'":': 4, '<': 11, '<?xml version=': 0, '>': 20, 'https://': 4, 'lorem ipsum': 0, 'www.': 4, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4604215', 'n_tokens_mistral': 2143, 'n_tokens_neox': 2109, 'n_words': 774}