<commit_before>#!/usr/bin/env python
# -*- encoding: utf-8 -*-
import argparse
import os
import subprocess
import sys
import time
import traceback
from typing import Any, Dict, List, Optional

import OpenSSL
import pem
from dns import resolver
from lexicon.client import Client
from lexicon.config import ConfigResolver

from dnsrobocert.core import config, utils


def main(args: List[str] = None) -> int:
    if not args:
        args = sys.argv[1:]

    parser = argparse.ArgumentParser()
    parser.add_argument(
        "-t", "--type", choices=["auth", "cleanup", "deploy"], required=True
    )
    parser.add_argument("-c", "--config", required=True)
    parser.add_argument("-l", "--lineage", default="None", required=False)

    parsed_args = parser.parse_args(args)
    dnsrobocert_config = config.load(parsed_args.config)

    if not dnsrobocert_config:
        print(
            "Error occured while loading the configuration file, aborting the `{0}` hook.".format(
                parsed_args.type
            ),
            file=sys.stderr,
        )
        return 1

    try:
        globals()[parsed_args.type](dnsrobocert_config, parsed_args.lineage)
    except BaseException as e:
        print(
            "Error while executing the `{0}` hook:".format(parsed_args.type),
            file=sys.stderr,
        )
        print(e, file=sys.stderr)
        traceback.print_exc(file=sys.stderr)
        return 1

    return 0


def auth(dnsrobocert_config: Dict[str, Any], lineage: str):
    profile = config.find_profile_for_lineage(dnsrobocert_config, lineage)
    domain = os.environ["CERTBOT_DOMAIN"]
    token = os.environ["CERTBOT_VALIDATION"]

    print("Executing auth hook for domain {0}, lineage {1}.".format(domain, lineage))

    _txt_challenge(profile, token, domain, action="create")

    remaining_challenges = int(os.environ.get("CERTBOT_REMAINING_CHALLENGES", "0"))
    if remaining_challenges != 0:
        print(
            "Still {0} challenges to handle, skip checks until last challenge.".format(
                remaining_challenges
            )
        )
        return

    all_domains = os.environ.get("CERTBOT_ALL_DOMAINS", "")
    all_domains = all_domains.split(",")
    challenges_to_check = [
        "_acme-challenge.{0}".format(domain) for domain in all_domains
    ]

    sleep_time = profile.get("sleep_time", 30)
    max_checks = profile.get("max_checks", 0)
    if max_checks:
        print("Challenges to check: {0}".format(challenges_to_check))
        checks = 0
        while True:
            checks = checks + 1
            if checks > max_checks:
                print(
                    "All challenges were not propagated after the maximum tries of {0}".format(
                        max_checks
                    ),
                    file=sys.stderr,
                )
                raise RuntimeError("Auth hook failed.")

            print(
                "Wait {0} seconds before checking that all challenges have the expected value "
                "(try {1}/{2})".format(sleep_time, checks, max_checks)
            )
            time.sleep(sleep_time)

            challenges_to_check = [
                challenge
                for challenge in challenges_to_check
                if not _check_one_challenge(
                    challenge,
                    token
                    if challenge == "_acme-challenge.{0}".format(domain)
                    else None,
                )
            ]

            if not challenges_to_check:
                print(
                    "All challenges have been propagated (try {0}/{1}).".format(
                        checks, max_checks
                    )
                )
                break
    else:
        print(
            "Wait {0} seconds to let all challenges be propagated: {1}".format(
                sleep_time, challenges_to_check
            )
        )
        time.sleep(sleep_time)


def _check_one_challenge(challenge: str, token: Optional[str]) -> bool:
    try:
        answers = resolver.query(challenge, "TXT")
    except (resolver.NXDOMAIN, resolver.NoAnswer) as e:
        print("TXT {0} does not exist.".format(challenge))
        return False<commit_msg>Fix quality<commit_after>#!/usr/bin/env python
# -*- encoding: utf-8 -*-
import argparse
import os
import subprocess
import sys
import time
import traceback
from typing import Any, Dict, List, Optional

import OpenSSL
import pem
from dns import resolver
from lexicon.client import Client
from lexicon.config import ConfigResolver

from dnsrobocert.core import config, utils


def main(args: List[str] = None) -> int:
    if not args:
        args = sys.argv[1:]

    parser = argparse.ArgumentParser()
    parser.add_argument(
        "-t", "--type", choices=["auth", "cleanup", "deploy"], required=True
    )
    parser.add_argument("-c", "--config", required=True)
    parser.add_argument("-l", "--lineage", default="None", required=False)

    parsed_args = parser.parse_args(args)
    dnsrobocert_config = config.load(parsed_args.config)

    if not dnsrobocert_config:
        print(
            "Error occured while loading the configuration file, aborting the `{0}` hook.".format(
                parsed_args.type
            ),
            file=sys.stderr,
        )
        return 1

    try:
        globals()[parsed_args.type](dnsrobocert_config, parsed_args.lineage)
    except BaseException as e:
        print(
            "Error while executing the `{0}` hook:".format(parsed_args.type),
            file=sys.stderr,
        )
        print(e, file=sys.stderr)
        traceback.print_exc(file=sys.stderr)
        return 1

    return 0


def auth(dnsrobocert_config: Dict[str, Any], lineage: str):
    profile = config.find_profile_for_lineage(dnsrobocert_config, lineage)
    domain = os.environ["CERTBOT_DOMAIN"]
    token = os.environ["CERTBOT_VALIDATION"]

    print("Executing auth hook for domain {0}, lineage {1}.".format(domain, lineage))

    _txt_challenge(profile, token, domain, action="create")

    remaining_challenges = int(os.environ.get("CERTBOT_REMAINING_CHALLENGES", "0"))
    if remaining_challenges != 0:
        print(
            "Still {0} challenges to handle, skip checks until last challenge.".format(
                remaining_challenges
            )
        )
        return

    all_domains_str = os.environ.get("CERTBOT_ALL_DOMAINS", "")
    all_domains = all_domains_str.split(",")
    challenges_to_check = [
        "_acme-challenge.{0}".format(domain) for domain in all_domains
    ]

    sleep_time = profile.get("sleep_time", 30)
    max_checks = profile.get("max_checks", 0)
    if max_checks:
        print("Challenges to check: {0}".format(challenges_to_check))
        checks = 0
        while True:
            checks = checks + 1
            if checks > max_checks:
                print(
                    "All challenges were not propagated after the maximum tries of {0}".format(
                        max_checks
                    ),
                    file=sys.stderr,
                )
                raise RuntimeError("Auth hook failed.")

            print(
                "Wait {0} seconds before checking that all challenges have the expected value "
                "(try {1}/{2})".format(sleep_time, checks, max_checks)
            )
            time.sleep(sleep_time)

            challenges_to_check = [
                challenge
                for challenge in challenges_to_check
                if not _check_one_challenge(
                    challenge,
                    token
                    if challenge == "_acme-challenge.{0}".format(domain)
                    else None,
                )
            ]

            if not challenges_to_check:
                print(
                    "All challenges have been propagated (try {0}/{1}).".format(
                        checks, max_checks
                    )
                )
                break
    else:
        print(
            "Wait {0} seconds to let all challenges be propagated: {1}".format(
                sleep_time, challenges_to_check
            )
        )
        time.sleep(sleep_time)


def _check_one_challenge(challenge: str, token: Optional[str]) -> bool:
    try:
        answers = resolver.query(challenge, "TXT")
    except (resolver.NXDOMAIN, resolver.NoAnswer):
        print("TXT {0} does not exist.".format(challenge))
        return False
{'fraction_non_alphanumeric': 0.10055601561575772, 'fraction_numerical': 0.006861469300839938, 'mean_word_length': 1.9746657283603097, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, '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': '3719283', 'n_tokens_mistral': 2533, 'n_tokens_neox': 2368, 'n_words': 588}