File: async_basic_user_mfa.py

package info (click to toggle)
python-duo-client 5.5.0-1
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid, trixie
  • size: 768 kB
  • sloc: python: 7,105; sh: 6; makefile: 4
file content (95 lines) | stat: -rw-r--r-- 3,410 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
"""
Example of Duo Auth API user authentication using asynchronous resquest/response methods
"""

import duo_client
import sys
import getpass


argv_iter = iter(sys.argv[1:])


def _get_next_arg(prompt, secure=False):
    """Read information from STDIN, using getpass when sensitive information should not be echoed to tty"""
    try:
        return next(argv_iter)
    except StopIteration:
        if secure is True:
            return getpass.getpass(prompt)
        else:
            return input(prompt)


def prompt_for_credentials() -> dict:
    """Collect required API credentials from command line prompts

    :return: dictionary containing Duo Auth API ikey, skey and hostname strings
    """

    ikey = _get_next_arg('Duo Auth API integration key ("DI..."): ')
    skey = _get_next_arg('Duo Auth API integration secret key: ', secure=True)
    host = _get_next_arg('Duo Auth API hostname ("api-....duosecurity.com"): ')
    username = _get_next_arg('Duo Username: ')

    return {"USERNAME": username, "IKEY": ikey, "SKEY": skey, "APIHOST": host}


def main():
    """Main program entry point"""

    inputs = prompt_for_credentials()

    auth_client = duo_client.Auth(
            ikey=inputs['IKEY'],
            skey=inputs['SKEY'],
            host=inputs['APIHOST']
    )

    # Verify that the Duo service is available
    duo_ping = auth_client.ping()
    if 'time' in duo_ping:
        print("\nDuo service check completed successfully.")
    else:
        print(f"Error: {duo_ping}")

    # Verify that IKEY and SKEY information provided are valid
    duo_check= auth_client.check()
    if 'time' in duo_check:
        print("IKEY and SKEY provided have been verified.")
    else:
        print(f"Error: {duo_check}")

    # Execute pre-authentication for given user
    print(f"\nExecuting pre-authentication for {inputs['USERNAME']}...")
    pre_auth = auth_client.preauth(username=inputs['USERNAME'])

    if pre_auth['result'] == "auth":
        try:
            print(f"Executing authentication action for {inputs['USERNAME']}...")
            auth = auth_client.auth(factor="push", username=inputs['USERNAME'], device="auto", async_txn=True)
            if 'txid' in auth:
                waiting = True
                # Collect the authentication result
                print("Getting authentication result...")
                # Repeat long polling for async authentication status until no longer in a 'waiting' state
                while waiting is True:
                    # Poll Duo Auth API for the status of the async authentication based upon transaction ID
                    auth_status = auth_client.auth_status(auth['txid'])
                    print(f"Auth status: {auth_status}")
                    if auth_status['waiting'] is not True:
                        # Waiting for response too async authentication is no longer 'True', so break the loop
                        waiting = False
                # Parse response for the 'status' dictionary key to determine whether to allow or deny
                print(auth_status)
            else:
                # Some kind of unexpected error occurred
                print(f"Error: an unknown error occurred attempting authentication for [{inputs['USERNAME']}]")
        except Exception as e_str:
            print(e_str)
    else:
        print(pre_auth['status_msg'])


if __name__ == '__main__':
    main()