Python腳本實(shí)現(xiàn)自動化管理Windows域

更新時間：2026年05月07日 09:09:18 作者：紅魔Y

這篇文章主要為大家詳細(xì)介紹了如何通過Python 的 ldap3 和 PowerShell 實(shí)現(xiàn)域用戶批量管理,組策略自動推送等操作,文中的示例代碼講解詳細(xì),感興趣的小伙伴可以了解下

為什么需要自動化域管理？

Active Directory (AD) 是企業(yè) IT 基礎(chǔ)設(shè)施的核心，幾乎所有管理工作都圍繞它展開：

操作	手動方式	Python 自動化
創(chuàng)建用戶	ADUC 里右鍵→新建→填表	一行腳本
重置密碼	找到用戶→右鍵→重置→輸入	批量腳本
查找鎖定賬戶	搜索→篩選→逐個查看	一鍵列出
創(chuàng)建計算機(jī)賬戶	ADUC→ Computers→新建	批量導(dǎo)入
推送組策略	GPMC→找到策略→鏈接	腳本部署
軟件部署	組策略→軟件安裝→配置	SCCM/PS 腳本

方案一：ldap3 庫（純 Python 方案）

ldap3 是一個純 Python 實(shí)現(xiàn)的 LDAP 客戶端，不需要安裝任何 C 擴(kuò)展。

安裝

pip install ldap3

連接 AD 域控制器

from ldap3 import Server, Connection, ALL, SUBTREE, MODIFY_ADD, MODIFY_REPLACE
from ldap3.core.exceptions import LDAPException

class ADManager:
    """Active Directory 域管理器"""

    def __init__(self, domain, dc_host, username, password, use_ssl=True):
        """
        domain: 例如 "corp.example.com"
        dc_host: 域控制器地址，例如 "dc01.corp.example.com"
        username: 管理員賬號，例如 "admin@corp.example.com"
        password: 密碼
        """
        self.domain = domain
        self.server = Server(
            dc_host,
            use_ssl=use_ssl,
            get_info=ALL,
        )
        self.conn = Connection(
            self.server,
            user=username,
            password=password,
            authentication="NTLM",
            auto_bind=True,
        )

        if self.conn.bind():
            print(f"? 已連接到域控制器: {dc_host}")
        else:
            raise LDAPException(f"連接失敗: {self.conn.last_error}")

        # 基礎(chǔ) DN
        self.base_dn = ",".join(
            f"DC={part}" for part in domain.split(".")
        )

    def search_users(self, filter_expr=None, attributes=None):
        """
        搜索域用戶
        filter_expr: LDAP 過濾表達(dá)式
        attributes: 返回的屬性列表
        """
        if filter_expr is None:
            filter_expr = "(objectClass=user)"

        if attributes is None:
            attributes = [
                "cn", "sAMAccountName", "mail",
                "department", "title", "whenCreated",
                "lastLogon", "userAccountControl",
                "memberOf", "distinguishedName",
            ]

        self.conn.search(
            search_base=self.base_dn,
            search_filter=filter_expr,
            search_scope=SUBTREE,
            attributes=attributes,
        )

        users = []
        for entry in self.conn.entries:
            user = {}
            for attr in attributes:
                try:
                    value = getattr(entry, attr, None)
                    user[attr] = str(value.value) if value else ""
                except Exception:
                    user[attr] = ""
            users.append(user)

        return users

    def close(self):
        """關(guān)閉連接"""
        self.conn.unbind()


# 使用
ad = ADManager(
    domain="corp.example.com",
    dc_host="dc01.corp.example.com",
    username="admin@corp.example.com",
    password="YourPassword"
)

# 搜索所有用戶
users = ad.search_users()
print(f"域中共有 {len(users)} 個用戶\n")

# 搜索特定部門
dev_users = ad.search_users(
    filter_expr="(&(objectClass=user)(department=Engineering))"
)
print(f"Engineering 部門: {len(dev_users)} 人")

ad.close()

用戶管理 CRUD

class ADUserManager(ADManager):
    """AD 用戶管理擴(kuò)展"""

    def create_user(
        self,
        username,
        first_name,
        last_name,
        password,
        ou_path=None,
        department="",
        title="",
        email="",
        groups=None
    ):
        """
        創(chuàng)建域用戶
        username: 登錄名 (sAMAccountName)
        ou_path: OU 路徑，例如 "OU=Users,OU=Shanghai,DC=corp,DC=example,DC=com"
        """
        if ou_path is None:
            ou_path = f"CN=Users,{self.base_dn}"

        dn = f"CN={username},{ou_path}"
        display_name = f"{first_name} {last_name}"

        # UPN (User Principal Name)
        upn = f"{username}@{self.domain}"

        # 默認(rèn)密碼需要符合策略
        default_password = password or "P@ssw0rd123!"

        try:
            self.conn.add(
                dn,
                attributes={
                    "objectClass": ["top", "person", "organizationalPerson", "user"],
                    "cn": username,
                    "sAMAccountName": username,
                    "userPrincipalName": upn,
                    "displayName": display_name,
                    "givenName": first_name,
                    "sn": last_name,
                    "mail": email or f"{username}@{self.domain}",
                    "department": department,
                    "title": title,
                    "userAccountControl": 512,  # 512 = 正常啟用
                }
            )

            if self.conn.result["result"] == 0:
                print(f"? 用戶創(chuàng)建成功: {username}")
                # 設(shè)置密碼
                self._set_password(dn, default_password)
                # 加入組
                if groups:
                    for group in groups:
                        self.add_to_group(username, group)
                return True
            else:
                print(f"? 創(chuàng)建失敗: {self.conn.result['description']}")
                return False

        except LDAPException as e:
            print(f"? 創(chuàng)建失敗: {e}")
            return False

    def _set_password(self, user_dn, new_password):
        """設(shè)置用戶密碼"""
        try:
            # 注意：設(shè)置密碼需要 SSL/TLS 連接
            self.conn.extend.microsoft.modify_password(
                user_dn, new_password
            )
            print(f"  密碼已設(shè)置")
            return True
        except Exception as e:
            print(f"  密碼設(shè)置失敗: {e}")
            return False

    def reset_password(self, username, new_password):
        """重置用戶密碼"""
        user_dn = self._find_user_dn(username)
        if user_dn:
            return self._set_password(user_dn, new_password)
        return False

    def _find_user_dn(self, username):
        """根據(jù)用戶名查找 DN"""
        self.conn.search(
            self.base_dn,
            f"(sAMAccountName={username})",
            attributes=["distinguishedName"],
        )
        if self.conn.entries:
            return str(self.conn.entries[0].distinguishedName)
        return None

    def disable_user(self, username):
        """禁用用戶賬戶"""
        user_dn = self._find_user_dn(username)
        if not user_dn:
            print(f"用戶不存在: {username}")
            return False

        # 514 = 禁用
        self.conn.modify(
            user_dn,
            {"userAccountControl": [(MODIFY_REPLACE, [514])]}
        )
        if self.conn.result["result"] == 0:
            print(f"? 已禁用: {username}")
            return True
        return False

    def enable_user(self, username):
        """啟用用戶賬戶"""
        user_dn = self._find_user_dn(username)
        if not user_dn:
            return False

        # 512 = 正常啟用
        self.conn.modify(
            user_dn,
            {"userAccountControl": [(MODIFY_REPLACE, [512])]}
        )
        if self.conn.result["result"] == 0:
            print(f"? 已啟用: {username}")
            return True
        return False

    def add_to_group(self, username, group_name):
        """將用戶添加到組"""
        user_dn = self._find_user_dn(username)
        if not user_dn:
            return False

        group_dn = self._find_group_dn(group_name)
        if not group_dn:
            print(f"組不存在: {group_name}")
            return False

        self.conn.modify(
            group_dn,
            {"member": [(MODIFY_ADD, [user_dn])]}
        )
        if self.conn.result["result"] == 0:
            print(f"  ? {username} 已加入 {group_name}")
            return True
        print(f"  ? 加入組失敗: {self.conn.result['description']}")
        return False

    def _find_group_dn(self, group_name):
        """根據(jù)組名查找 DN"""
        self.conn.search(
            self.base_dn,
            f"(&(objectClass=group)(cn={group_name}))",
            attributes=["distinguishedName"],
        )
        if self.conn.entries:
            return str(self.conn.entries[0].distinguishedName)
        return None


# 使用示例
# mgr = ADUserManager("corp.example.com", "dc01.corp.example.com", ...)
#
# 批量創(chuàng)建用戶
# new_users = [
#     {"username": "zhangsan", "first_name": "三", "last_name": "張", "department": "Engineering", "groups": ["VPN-Users", "Dev-Team"]},
#     {"username": "lisi", "first_name": "四", "last_name": "李", "department": "HR", "groups": ["HR-Team"]},
# ]
# for u in new_users:
#     mgr.create_user(**u)

批量操作：從 CSV 導(dǎo)入

import csv

def bulk_create_users_from_csv(ad_manager, csv_file, default_password="P@ssw0rd123!"):
    """從 CSV 文件批量創(chuàng)建用戶"""
    with open(csv_file, "r", encoding="utf-8-sig") as f:
        reader = csv.DictReader(f)

        results = {"success": 0, "failed": 0, "errors": []}

        for row in reader:
            username = row.get("username", "").strip()
            if not username:
                continue

            groups = [g.strip() for g in row.get("groups", "").split(",") if g.strip()]

            success = ad_manager.create_user(
                username=username,
                first_name=row.get("first_name", ""),
                last_name=row.get("last_name", ""),
                password=default_password,
                department=row.get("department", ""),
                title=row.get("title", ""),
                groups=groups if groups else None,
            )

            if success:
                results["success"] += 1
            else:
                results["failed"] += 1
                results["errors"].append(username)

    print(f"\n批量創(chuàng)建完成: {results['success']} 成功, {results['failed']} 失敗")
    if results["errors"]:
        print(f"失敗用戶: {', '.join(results['errors'][:10])}")

    return results


# CSV 文件格式示例:
# username,first_name,last_name,department,title,groups
# zhangsan,三,張,Engineering,Developer,"VPN-Users,Dev-Team"
# lisi,四,李,HR,HR Specialist,HR-Team

方案二：PowerShell 橋接（無需 ldap3）

如果無法安裝 ldap3，通過 PowerShell 的 ActiveDirectory 模塊同樣可以實(shí)現(xiàn)：

import subprocess
import json

class ADPowerShell:
    """通過 PowerShell 管理 AD"""

    def __init__(self):
        # 檢查 ActiveDirectory 模塊是否可用
        result = subprocess.run(
            ["powershell", "-Command",
             "Get-Module -ListAvailable ActiveDirectory"],
            capture_output=True, text=True
        )
        if not result.stdout.strip():
            print("?? ActiveDirectory 模塊未安裝")
            print("請安裝 RSAT (Remote Server Administration Tools)")

    def get_all_users(self, properties=None):
        """獲取所有用戶"""
        if properties is None:
            properties = "Name,SamAccountName,Department,Title,Enabled,LastLogonDate"

        ps_cmd = f'Get-ADUser -Filter * -Properties {properties} | Select-Object {properties} | ConvertTo-Json -Depth 3'

        result = subprocess.run(
            ["powershell", "-Command", ps_cmd],
            capture_output=True, text=True, timeout=60
        )

        try:
            data = json.loads(result.stdout)
            return data if isinstance(data, list) else [data]
        except json.JSONDecodeError:
            return []

    def create_user(self, username, display_name, ou="", department="", groups=None):
        """創(chuàng)建用戶"""
        ou_part = f'-Path "{ou}"' if ou else ""

        cmd = f'''
        New-ADUser -Name "{display_name}" -SamAccountName "{username}" -UserPrincipalName "{username}@corp.example.com" -DisplayName "{display_name}" -Department "{department}" -AccountPassword (ConvertTo-SecureString "P@ssw0rd123!" -AsPlainText -Force) -Enabled $true {ou_part}
        '''

        result = subprocess.run(
            ["powershell", "-Command", cmd],
            capture_output=True, text=True
        )

        if result.returncode == 0:
            print(f"? 創(chuàng)建用戶: {username}")

            # 加入組
            if groups:
                for group in groups.split(","):
                    self.add_to_group(username, group.strip())
            return True
        else:
            print(f"? 創(chuàng)建失敗: {result.stderr}")
            return False

    def add_to_group(self, username, group_name):
        """將用戶添加到組"""
        cmd = f'Add-ADGroupMember -Identity "{group_name}" -Members "{username}"'

        result = subprocess.run(
            ["powershell", "-Command", cmd],
            capture_output=True, text=True
        )

        if result.returncode == 0:
            print(f"  ? 加入組: {group_name}")
            return True
        print(f"  ? 加入組失敗: {group_name}")
        return False

    def find_locked_users(self):
        """查找被鎖定的用戶"""
        ps_cmd = '''
        Search-ADAccount -LockedOut | Select-Object SamAccountName,Name,LastLogonDate,LockoutTime | ConvertTo-Json -Depth 3
        '''

        result = subprocess.run(
            ["powershell", "-Command", ps_cmd],
            capture_output=True, text=True
        )

        try:
            data = json.loads(result.stdout)
            users = data if isinstance(data, list) else [data]
            print(f"\n?? 被鎖定的用戶 ({len(users)} 個)：")
            for u in users:
                print(f"  {u['SamAccountName']} ({u['Name']})")
            return users
        except json.JSONDecodeError:
            print("無被鎖定的用戶")
            return []

    def find_inactive_users(self, days=90):
        """查找長時間未登錄的用戶"""
        ps_cmd = f'''
        Search-ADAccount -AccountInactive -TimeSpan {days} | Where-Object {{ $_.Enabled -eq $true }} | Select-Object SamAccountName,Name,LastLogonDate,DistinguishedName | ConvertTo-Json -Depth 3
        '''

        result = subprocess.run(
            ["powershell", "-Command", ps_cmd],
            capture_output=True, text=True, timeout=60
        )

        try:
            data = json.loads(result.stdout)
            users = data if isinstance(data, list) else [data]
            print(f"\n?? 超過 {days} 天未登錄的用戶 ({len(users)} 個)：")
            for u in users:
                last_logon = u.get("LastLogonDate", "從未登錄")
                print(f"  {u['SamAccountName']} | 最后登錄: {last_logon}")
            return users
        except json.JSONDecodeError:
            return []

    def unlock_user(self, username):
        """解鎖用戶"""
        cmd = f'Unlock-ADAccount -Identity "{username}"'
        result = subprocess.run(
            ["powershell", "-Command", cmd],
            capture_output=True, text=True
        )
        if result.returncode == 0:
            print(f"? 已解鎖: {username}")
            return True
        print(f"? 解鎖失敗: {username}")
        return False

    def reset_password(self, username, new_password):
        """重置密碼"""
        cmd = f'''
        Set-ADAccountPassword -Identity "{username}" -Reset -NewPassword (ConvertTo-SecureString "{new_password}" -AsPlainText -Force)
        '''
        result = subprocess.run(
            ["powershell", "-Command", cmd],
            capture_output=True, text=True
        )
        if result.returncode == 0:
            print(f"? 密碼已重置: {username}")
            # 強(qiáng)制下次登錄修改密碼
            self.conn = subprocess.run(
                ["powershell", "-Command", f'Set-ADUser -Identity "{username}" -ChangePasswordAtLogon $true'],
                capture_output=True, text=True
            )
            return True
        print(f"? 密碼重置失敗: {username}")
        return False


# 使用
# ad = ADPowerShell()
# ad.find_locked_users()
# ad.find_inactive_users(days=90)
# ad.unlock_user("zhangsan")
# ad.reset_password("zhangsan", "NewP@ssw0rd!")

實(shí)戰(zhàn)：組策略自動化

def deploy_software_via_gpo(
    gpo_name,
    software_path,
    gpo_target_ou
):
    """
    通過組策略部署軟件
    gpo_name: GPO 名稱
    software_path: MSI 或 EXE 安裝包路徑
    gpo_target_ou: 目標(biāo) OU
    """
    ps_cmd = f'''
    # 創(chuàng)建新的 GPO
    $gpo = New-GPO -Name "{gpo_name}"

    # 配置軟件安裝
    # 方法：使用 GroupPolicy 模塊
    Import-Module GroupPolicy

    # 將 GPO 鏈接到目標(biāo) OU
    New-GPLink -Name "{gpo_name}" -Target "{gpo_target_ou}" -LinkEnabled Yes

    Write-Output "GPO '{gpo_name}' 已創(chuàng)建并鏈接到 '{gpo_target_ou}'"
    '''

    result = subprocess.run(
        ["powershell", "-Command", ps_cmd],
        capture_output=True, text=True
    )

    print(result.stdout)
    if result.returncode == 0:
        print("? GPO 部署成功")
        return True
    print(f"? 部署失敗: {result.stderr}")
    return False


def list_gpos():
    """列出所有組策略"""
    ps_cmd = 'Get-GPO -All | Select-Object DisplayName,Id,GpoStatus,CreationTime | ConvertTo-Json -Depth 3'

    result = subprocess.run(
        ["powershell", "-Command", ps_cmd],
        capture_output=True, text=True
    )

    try:
        data = json.loads(result.stdout)
        gpos = data if isinstance(data, list) else [data]
        print(f"\n組策略列表 ({len(gpos)} 個)：\n")
        for gpo in gpos:
            status = gpo.get("GpoStatus", "")
            print(f"  {gpo['DisplayName']} | {status} | {gpo.get('CreationTime', '')[:10]}")
        return gpos
    except json.JSONDecodeError:
        return []

實(shí)戰(zhàn)：計算機(jī)賬戶管理

def get_computer_inventory(ad_conn=None, ps_method=True):
    """獲取域內(nèi)計算機(jī)清單"""
    ps_cmd = '''
    Get-ADComputer -Filter * -Properties Name,OperatingSystem,OperatingSystemVersion,
        LastLogonDate,IPv4Address,Enabled |
    Select-Object Name,OperatingSystem,OperatingSystemVersion,
        LastLogonDate,IPv4Address,Enabled |
    ConvertTo-Json -Depth 3
    '''

    result = subprocess.run(
        ["powershell", "-Command", ps_cmd],
        capture_output=True, text=True, timeout=60
    )

    try:
        data = json.loads(result.stdout)
        computers = data if isinstance(data, list) else [data]

        print(f"\n域內(nèi)計算機(jī)清單 ({len(computers)} 臺)：\n")
        print(f"{'計算機(jī)名':<20}{'操作系統(tǒng)':<25}{'最后登錄':<12}{'IP地址':<16}{'狀態(tài)'}")
        print("-" * 90)

        os_counter = Counter()
        for c in computers:
            name = c.get("Name", "")
            os_name = c.get("OperatingSystem", "")
            os_version = c.get("OperatingSystemVersion", "")
            os_counter[os_name] += 1
            last_logon = str(c.get("LastLogonDate", "從未"))[:10]
            ip = c.get("IPv4Address", "")
            enabled = "?" if c.get("Enabled") else "?"
            print(f"{name:<20}{os_name:<25}{last_logon:<12}{ip:<16}{enabled}")

        print(f"\n操作系統(tǒng)分布：")
        for os_name, count in os_counter.most_common():
            print(f"  {os_name}: {count} 臺")

        return computers
    except json.JSONDecodeError:
        return []


from collections import Counter

# 使用
# get_computer_inventory()

實(shí)戰(zhàn)：安全審計腳本

def security_audit():
    """AD 安全審計"""
    print("=" * 60)
    print("  Active Directory 安全審計報告")
    print("=" * 60)

    # 1. 域管理員賬戶
    print("\n=== 域管理員賬戶 ===")
    ps_cmd = 'Get-ADGroupMember -Identity "Domain Admins" | Select-Object Name,SamAccountName | ConvertTo-Json'
    result = subprocess.run(["powershell", "-Command", ps_cmd], capture_output=True, text=True)
    try:
        admins = json.loads(result.stdout)
        admins = admins if isinstance(admins, list) else [admins]
        print(f"域管理員數(shù)量: {len(admins)}")
        for a in admins:
            print(f"  - {a.get('SamAccountName', a.get('Name', 'Unknown'))}")
    except json.JSONDecodeError:
        pass

    # 2. 被鎖定的賬戶
    print("\n=== 被鎖定賬戶 ===")
    ps_cmd = 'Search-ADAccount -LockedOut | Select-Object SamAccountName | ConvertTo-Json'
    result = subprocess.run(["powershell", "-Command", ps_cmd], capture_output=True, text=True)
    try:
        locked = json.loads(result.stdout)
        locked = locked if isinstance(locked, list) else [locked]
        print(f"鎖定賬戶: {len(locked)}")
        for a in locked:
            print(f"  - {a.get('SamAccountName')}")
    except json.JSONDecodeError:
        print("無鎖定賬戶")

    # 3. 密碼永不過期的賬戶
    print("\n=== 密碼永不過期的賬戶 ===")
    ps_cmd = 'Get-ADUser -Filter * -Properties PasswordNeverExpires | Where-Object {$_.PasswordNeverExpires -eq $true -and $_.Enabled -eq $true} | Select-Object SamAccountName | ConvertTo-Json'
    result = subprocess.run(["powershell", "-Command", ps_cmd], capture_output=True, text=True, timeout=30)
    try:
        never_expire = json.loads(result.stdout)
        never_expire = never_expire if isinstance(never_expire, list) else [never_expire]
        print(f"永不過期賬戶: {len(never_expire)}")
        for a in never_expire[:20]:
            print(f"  - {a.get('SamAccountName')}")
        if len(never_expire) > 20:
            print(f"  ... 還有 {len(never_expire) - 20} 個")
    except json.JSONDecodeError:
        print("無（或查詢失敗）")

    # 4. 不活躍賬戶
    print("\n=== 90天未登錄的活躍賬戶 ===")
    ps_cmd = 'Search-ADAccount -AccountInactive -TimeSpan 90 | Where-Object {$_.Enabled -eq $true} | Select-Object SamAccountName,LastLogonDate | ConvertTo-Json -Depth 3'
    result = subprocess.run(["powershell", "-Command", ps_cmd], capture_output=True, text=True, timeout=30)
    try:
        inactive = json.loads(result.stdout)
        inactive = inactive if isinstance(inactive, list) else [inactive]
        print(f"不活躍賬戶: {len(inactive)}")
        for a in inactive[:10]:
            print(f"  - {a.get('SamAccountName')} (最后登錄: {a.get('LastLogonDate', '從未')})")
    except json.JSONDecodeError:
        print("無（或查詢失?。?)

    print("\n" + "=" * 60)


# 使用（需要管理員權(quán)限）
# security_audit()

小結(jié)

需求	方案	模塊
查詢用戶	ldap3 / PowerShell	`search_users` / `Get-ADUser`
創(chuàng)建用戶	ldap3 / PowerShell	`conn.add` / `New-ADUser`
重置密碼	ldap3 / PowerShell	`modify_password` / `Set-ADAccountPassword`
解鎖賬戶	PowerShell	`Unlock-ADAccount`
組管理	ldap3 / PowerShell	`MODIFY_ADD` / `Add-ADGroupMember`
安全審計	PowerShell	`Search-ADAccount`
軟件部署	GPO	`New-GPO` + `New-GPLink`
批量導(dǎo)入	CSV + 循環(huán)	`bulk_create_users_from_csv`