目的

現在のActive Directoryのユーザー情報をPythonで取得しようと思い、
ldap3を利用して、Active Directoryのユーザー情報を取得してみました。

環境

Windows 10 x64 1809
Python 3.6.5 x64
Power Shell 6 x64
Visual Studio Code x64
Git for Windows x64

ldap3を利用したユーザー情報取得の流れ

PythonでActive Directoryを参照するにはldap3が良さそうです。

> pip install ldap3

まずは、ldap3をインポートして、Serverインスタンスを生成してみます。

from ldap3 import Server, Connection, ALL, NTLM, ALL_ATTRIBUTES, ALL_OPERATIONAL_ATTRIBUTES, AUTO_BIND_NO_TLS, SUBTREE
from ldap3.core.exceptions import LDAPCursorError

server = Server('サーバー名', port = 389, get_info = ALL)

ポートは389固定、すべての情報を取得としています。

続いて、Active Directoryに接続してみます。

conn = Connection(server, user = 'ドメイン\\ユーザー',
                          'パスワード',
                          authentication = NTLM,
                          auto_bind = True)

authenticationをNTLMとしています。

そして、ユーザー情報を問い合わせてみます。

conn.search('DC=hoge,DC=loacl',
            '(&(|(objectclass=user)(objectclass=person)(objectclass=inetOrgPerson)(objectclass=organizationalPerson))(!(objectclass=computer)))',
            attributes = [ALL_ATTRIBUTES, ALL_OPERATIONAL_ATTRIBUTES])

問い合わせの対象をユーザー情報に限定しています。（コンピューターを除外）

あとは、取得した結果を利用します。

for entry in sorted(conn.entries):
    print(entry.sAMAccountName)

存在しない情報を参照するとLDAPCursorErrorが発生するので、適宜try/exceptを行います。

ldap3を利用したユーザー情報取得のコード

> pip install ldap3
> pip install python-dotenv

import sys
import csv
import settings
from ldap3 import Server, Connection, ALL, NTLM, ALL_ATTRIBUTES, ALL_OPERATIONAL_ATTRIBUTES, AUTO_BIND_NO_TLS, SUBTREE
from ldap3.core.exceptions import LDAPCursorError

class ADInfo:
    # コンストラクタ
    def __init__(self, server_name, domain_name, user_name, password, search_dc):
        self.__server_name = server_name
        self.__domain_name = domain_name
        self.__user_name = user_name
        self.__password = password
        self.__search_dc = search_dc

    # 接続
    def connect(self):
        self.__server = Server(self.__server_name, port = 389, get_info = ALL)
        self.__conn = Connection(self.__server,
                                 user = f'{self.__domain_name}\\{self.__user_name}',
                                 password = self.__password,
                                 authentication = NTLM,
                                 auto_bind = True)

    # 接続ユーザーを確認
    def who_am_i(self):
        return self.__conn.extend.standard.who_am_i()

    # ユーザー情報を取得
    def search_users(self):
        self.__conn.search(self.__search_dc,
                           '(&(|(objectclass=user)(objectclass=person)(objectclass=inetOrgPerson)(objectclass=organizationalPerson))(!(objectclass=computer)))',
                           attributes = [ALL_ATTRIBUTES, ALL_OPERATIONAL_ATTRIBUTES])
        return self.__conn.entries

    # CSVファイルに出力
    def output_users(self, entries, csv_filepath):
        with open(csv_filepath, 'w', encoding="cp932", errors="ignore", newline='') as f:
            writer = csv.writer(f)
            writer.writerow(['id', '氏名', '説明', '所属グループ'])
            for entry in sorted(entries):
                try:
                    desc = entry.description
                except LDAPCursorError:
                    desc = ""
                try:
                    memberOf = entry.memberOf
                except LDAPCursorError:
                    memberOf = [""]
                for group in memberOf:
                    group_cnname = group.split(",")[0].lstrip("CN=")
                    writer.writerow([entry.sAMAccountName, entry.name, desc, group_cnname])
                    # print([entry.sAMAccountName, entry.name, desc, group_cnname])

RETURN_SUCCESS = 0
RETURN_FAILURE = -1

def main():
    print("===================================================================")
    print("Active Directory のユーザー情報を取得してCSVファイルを出力します。")
    print("ユーザーごとに所属しているグループも取得します。")
    print("===================================================================")

    # 引数のチェック
    argvs = sys.argv
    if len(argvs) != 2 or not argvs[1]:
        print("CSVファイルのパスを指定してください。")
        return RETURN_FAILURE

    # CSVファイルパスの取得
    csv_filepath = argvs[1].strip()

    try:
        # AD情報
        adinfo = ADInfo(settings.SERVER_NAME,
                        settings.DOMAIN_NAME,
                        settings.USER_NAME,
                        settings.PASSWORD,
                        settings.SEARCH_DC)
        adinfo.connect()
        print(adinfo.who_am_i())

        # CSV出力
        entries = adinfo.search_users()
        adinfo.output_users(entries, csv_filepath)
    except Exception as e:
        print(e)
        return RETURN_FAILURE

    return RETURN_SUCCESS

if __name__ == "__main__":
    main()

設定情報はpython-dotenvを利用しています。

import os
from os.path import join, dirname
from dotenv import load_dotenv

dotenv_path = join(dirname(__file__), '.env')
load_dotenv(dotenv_path)

SERVER_NAME = os.environ.get("SERVER_NAME")
DOMAIN_NAME = os.environ.get("DOMAIN_NAME")
USER_NAME = os.environ.get("USER_NAME")
PASSWORD = os.environ.get("PASSWORD")
SEARCH_DC = os.environ.get("SEARCH_DC")

設定ファイル（.env）はこんな感じ

SERVER_NAME = 'IPorSERVERNAME'
DOMAIN_NAME = 'DOMAINNAME'
USER_NAME = 'USERNAME'
PASSWORD = 'PASSWORD'
SEARCH_DC = 'DC=xxxx,DC=xxxx'

まとめ

Pythonでユーザー情報を取得することができました。
いろいろな情報が取得できそうなので、ユーザーやグループ以外の情報も活用できそうです。

取得対象の件数が多い場合は、extend.standard.paged_searchメソッドでページ単位の検索をしたほうがよいようです。