This commit is contained in:
beabigegg
2025-09-01 16:42:41 +08:00
parent 22a231d78c
commit 00061adeb7
23 changed files with 858 additions and 3584 deletions

View File

@@ -1,141 +0,0 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""Create sample todo data for testing"""
import os
import sys
from dotenv import load_dotenv
import pymysql
from datetime import datetime, timedelta
import uuid
# Load environment variables
load_dotenv()
def create_sample_todos():
"""Create sample todo items for testing"""
print("=" * 60)
print("Creating Sample Todo Data")
print("=" * 60)
db_config = {
'host': os.getenv('MYSQL_HOST', 'mysql.theaken.com'),
'port': int(os.getenv('MYSQL_PORT', 33306)),
'user': os.getenv('MYSQL_USER', 'A060'),
'password': os.getenv('MYSQL_PASSWORD', 'WLeSCi0yhtc7'),
'database': os.getenv('MYSQL_DATABASE', 'db_A060'),
'charset': 'utf8mb4'
}
try:
connection = pymysql.connect(**db_config)
cursor = connection.cursor()
# Sample todos data
sample_todos = [
{
'title': '完成網站改版設計稿',
'description': '設計新版網站的主要頁面布局,包含首頁、產品頁面和聯絡頁面',
'status': 'DOING',
'priority': 'HIGH',
'due_date': (datetime.now() + timedelta(days=7)).date(),
'creator_ad': '92367',
'creator_display_name': 'ymirliu 陸一銘',
'creator_email': 'ymirliu@panjit.com.tw',
'starred': True
},
{
'title': '資料庫效能優化',
'description': '優化主要查詢語句,提升系統響應速度',
'status': 'NEW',
'priority': 'URGENT',
'due_date': (datetime.now() + timedelta(days=3)).date(),
'creator_ad': '92367',
'creator_display_name': 'ymirliu 陸一銘',
'creator_email': 'ymirliu@panjit.com.tw',
'starred': False
},
{
'title': 'API 文檔更新',
'description': '更新所有 API 介面文檔,補充新增的端點說明',
'status': 'DOING',
'priority': 'MEDIUM',
'due_date': (datetime.now() + timedelta(days=10)).date(),
'creator_ad': 'test',
'creator_display_name': '測試使用者',
'creator_email': 'test@panjit.com.tw',
'starred': False
},
{
'title': '使用者測試回饋整理',
'description': '整理上週使用者測試的所有回饋意見,並分類處理',
'status': 'BLOCKED',
'priority': 'LOW',
'due_date': (datetime.now() + timedelta(days=15)).date(),
'creator_ad': 'test',
'creator_display_name': '測試使用者',
'creator_email': 'test@panjit.com.tw',
'starred': True
},
{
'title': '系統安全性檢查',
'description': '對系統進行全面的安全性檢查,確保沒有漏洞',
'status': 'NEW',
'priority': 'URGENT',
'due_date': (datetime.now() + timedelta(days=2)).date(),
'creator_ad': '92367',
'creator_display_name': 'ymirliu 陸一銘',
'creator_email': 'ymirliu@panjit.com.tw',
'starred': False
}
]
created_count = 0
for todo in sample_todos:
todo_id = str(uuid.uuid4())
sql = """
INSERT INTO todo_item
(id, title, description, status, priority, due_date, created_at, creator_ad, creator_display_name, creator_email, starred)
VALUES
(%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
"""
values = (
todo_id,
todo['title'],
todo['description'],
todo['status'],
todo['priority'],
todo['due_date'],
datetime.now(),
todo['creator_ad'],
todo['creator_display_name'],
todo['creator_email'],
todo['starred']
)
cursor.execute(sql, values)
created_count += 1
print(f"[OK] Created todo: {todo['title']} (ID: {todo_id[:8]}...)")
connection.commit()
print(f"\n[SUCCESS] Created {created_count} sample todos successfully!")
# Show summary
cursor.execute("SELECT COUNT(*) FROM todo_item")
total_count = cursor.fetchone()[0]
print(f"[INFO] Total todos in database: {total_count}")
cursor.close()
connection.close()
return True
except Exception as e:
print(f"[ERROR] Failed to create sample data: {str(e)}")
return False
if __name__ == "__main__":
create_sample_todos()

View File

@@ -1,98 +0,0 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""Create test todos directly in database for testing public/private functionality"""
import pymysql
import uuid
import json
from datetime import datetime
import os
from dotenv import load_dotenv
# Load environment variables
load_dotenv()
def create_test_todos():
"""Create test todos directly in database"""
try:
# Connect to database
conn = pymysql.connect(
host=os.getenv('MYSQL_HOST', 'mysql.theaken.com'),
port=int(os.getenv('MYSQL_PORT', 33306)),
user=os.getenv('MYSQL_USER', 'A060'),
password=os.getenv('MYSQL_PASSWORD', 'WLeSCi0yhtc7'),
database=os.getenv('MYSQL_DATABASE', 'db_A060')
)
cursor = conn.cursor()
# Test data
todos = [
{
'id': str(uuid.uuid4()),
'title': '公開測試任務 - ymirliu 建立',
'description': '這是一個公開任務,其他人可以看到並追蹤',
'status': 'NEW',
'priority': 'MEDIUM',
'created_at': datetime.utcnow(),
'creator_ad': '92367',
'creator_display_name': 'ymirliu 劉念蒨',
'creator_email': 'ymirliu@panjit.com.tw',
'starred': False,
'is_public': True,
'tags': json.dumps(['測試', '公開功能'])
},
{
'id': str(uuid.uuid4()),
'title': '私人測試任務 - ymirliu 建立',
'description': '這是一個私人任務,只有建立者可見',
'status': 'NEW',
'priority': 'HIGH',
'created_at': datetime.utcnow(),
'creator_ad': '92367',
'creator_display_name': 'ymirliu 劉念蒨',
'creator_email': 'ymirliu@panjit.com.tw',
'starred': True,
'is_public': False,
'tags': json.dumps(['測試', '私人功能'])
}
]
# Insert todos
for todo in todos:
sql = """INSERT INTO todo_item
(id, title, description, status, priority, created_at, creator_ad,
creator_display_name, creator_email, starred, is_public, tags)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)"""
cursor.execute(sql, (
todo['id'], todo['title'], todo['description'], todo['status'],
todo['priority'], todo['created_at'], todo['creator_ad'],
todo['creator_display_name'], todo['creator_email'],
todo['starred'], todo['is_public'], todo['tags']
))
print(f"✓ 建立 {'公開' if todo['is_public'] else '私人'} Todo: {todo['title']}")
# Commit changes
conn.commit()
print(f"\n✅ 成功建立 {len(todos)} 個測試 Todo")
# Verify the insertion
cursor.execute("SELECT id, title, is_public FROM todo_item WHERE creator_ad = '92367'")
results = cursor.fetchall()
print(f"\n📊 資料庫中的測試數據:")
for result in results:
print(f" - {result[1]} ({'公開' if result[2] else '私人'})")
cursor.close()
conn.close()
except Exception as e:
print(f"❌ 建立測試數據失敗: {str(e)}")
return False
return True
if __name__ == "__main__":
create_test_todos()

View File

@@ -1,90 +0,0 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""Debug LDAP search to find the correct format"""
import os
import sys
from dotenv import load_dotenv
from ldap3 import Server, Connection, SUBTREE, ALL_ATTRIBUTES
# Load environment variables
load_dotenv()
def debug_ldap():
"""Debug LDAP search"""
print("=" * 60)
print("Debug LDAP Search")
print("=" * 60)
# Get LDAP configuration
ldap_server = os.getenv('LDAP_SERVER', 'ldap://panjit.com.tw')
ldap_port = int(os.getenv('LDAP_PORT', 389))
ldap_bind_user = os.getenv('LDAP_BIND_USER_DN', '')
ldap_bind_password = os.getenv('LDAP_BIND_USER_PASSWORD', '')
ldap_search_base = os.getenv('LDAP_SEARCH_BASE', 'DC=panjit,DC=com,DC=tw')
print(f"LDAP Server: {ldap_server}")
print(f"LDAP Port: {ldap_port}")
print(f"Search Base: {ldap_search_base}")
print("-" * 60)
try:
# Create server object
server = Server(
ldap_server,
port=ldap_port,
use_ssl=False,
get_info=ALL_ATTRIBUTES
)
# Create connection with bind user
conn = Connection(
server,
user=ldap_bind_user,
password=ldap_bind_password,
auto_bind=True,
raise_exceptions=True
)
print("[OK] Successfully connected to LDAP server")
# Test different search filters
test_searches = [
"(&(objectClass=person)(sAMAccountName=ymirliu))",
"(&(objectClass=person)(userPrincipalName=ymirliu@panjit.com.tw))",
"(&(objectClass=person)(mail=ymirliu@panjit.com.tw))",
"(&(objectClass=person)(cn=*ymirliu*))",
"(&(objectClass=person)(displayName=*ymirliu*))",
]
for i, search_filter in enumerate(test_searches, 1):
print(f"\n[{i}] Testing filter: {search_filter}")
conn.search(
ldap_search_base,
search_filter,
SUBTREE,
attributes=['sAMAccountName', 'displayName', 'mail', 'userPrincipalName', 'cn']
)
if conn.entries:
print(f" Found {len(conn.entries)} entries:")
for entry in conn.entries:
print(f" sAMAccountName: {entry.sAMAccountName}")
print(f" userPrincipalName: {entry.userPrincipalName}")
print(f" displayName: {entry.displayName}")
print(f" mail: {entry.mail}")
print(f" cn: {entry.cn}")
print()
else:
print(" No entries found")
conn.unbind()
except Exception as e:
print(f"[ERROR] LDAP connection failed: {str(e)}")
import traceback
traceback.print_exc()
if __name__ == "__main__":
debug_ldap()

View File

@@ -85,19 +85,19 @@ class TodoItem(db.Model):
def can_edit(self, user_ad):
"""Check if user can edit this todo"""
if self.creator_ad == user_ad:
return True
return any(r.ad_account == user_ad for r in self.responsible_users)
# Only creator can edit
return self.creator_ad == user_ad
def can_view(self, user_ad):
"""Check if user can view this todo"""
# Public todos can be viewed by anyone
if self.is_public:
return True
# Private todos can be viewed by creator, responsible users, and followers
if self.can_edit(user_ad):
# Private todos can be viewed by creator and responsible users only
if self.creator_ad == user_ad:
return True
return any(f.ad_account == user_ad for f in self.followers)
# Check if user is a responsible user
return any(r.ad_account == user_ad for r in self.responsible_users)
def can_follow(self, user_ad):
"""Check if user can follow this todo"""
@@ -154,7 +154,8 @@ class TodoAuditLog(db.Model):
actor_ad = db.Column(db.String(128), nullable=False)
todo_id = db.Column(CHAR(36), db.ForeignKey('todo_item.id', ondelete='SET NULL'))
action = db.Column(ENUM('CREATE', 'UPDATE', 'DELETE', 'COMPLETE', 'IMPORT',
'MAIL_SENT', 'MAIL_FAIL', 'FIRE_EMAIL', 'DIGEST_EMAIL', 'BULK_REMINDER'), nullable=False)
'MAIL_SENT', 'MAIL_FAIL', 'FIRE_EMAIL', 'DIGEST_EMAIL', 'BULK_REMINDER',
'FOLLOW', 'UNFOLLOW'), nullable=False)
detail = db.Column(JSON)
created_at = db.Column(db.DateTime, nullable=False, default=datetime.utcnow)

View File

@@ -142,11 +142,9 @@ def upload_excel():
if responsible_str and responsible_str != 'nan':
responsible_users = [user.strip() for user in responsible_str.replace(',', ';').split(';') if user.strip()]
# 追蹤人
followers_str = str(row.get('追蹤人', row.get('followers', ''))).strip()
followers = []
if followers_str and followers_str != 'nan':
followers = [user.strip() for user in followers_str.replace(',', ';').split(';') if user.strip()]
# 公開設定
is_public_str = str(row.get('公開設定', row.get('is_public', ''))).strip().lower()
is_public = is_public_str in ['', 'yes', 'true', '1', 'y'] if is_public_str and is_public_str != 'nan' else False
todos_data.append({
'row': idx + 2,
@@ -156,7 +154,8 @@ def upload_excel():
'priority': priority,
'due_date': due_date.isoformat() if due_date else None,
'responsible_users': responsible_users,
'followers': followers
'followers': [], # Excel模板中沒有followers欄位初始化為空陣列
'is_public': is_public
})
except Exception as e:
@@ -200,9 +199,8 @@ def import_todos():
for todo_data in todos_data:
try:
# 驗證負責人和追蹤人的 AD 帳號
# 驗證負責人的 AD 帳號
responsible_users = todo_data.get('responsible_users', [])
followers = todo_data.get('followers', [])
if responsible_users:
valid_responsible = validate_ad_accounts(responsible_users)
@@ -214,21 +212,17 @@ def import_todos():
})
continue
if followers:
valid_followers = validate_ad_accounts(followers)
invalid_followers = set(followers) - set(valid_followers.keys())
if invalid_followers:
errors.append({
'row': todo_data.get('row', '?'),
'error': f'無效的追蹤人帳號: {", ".join(invalid_followers)}'
})
continue
# 建立待辦事項
due_date = None
if todo_data.get('due_date'):
due_date = datetime.strptime(todo_data['due_date'], '%Y-%m-%d').date()
# 處理公開設定
is_public = False # 預設為非公開
if todo_data.get('is_public'):
is_public_str = str(todo_data['is_public']).strip().lower()
is_public = is_public_str in ['', 'yes', 'true', '1', 'y']
todo = TodoItem(
id=str(uuid.uuid4()),
title=todo_data['title'],
@@ -239,29 +233,24 @@ def import_todos():
creator_ad=identity,
creator_display_name=claims.get('display_name', identity),
creator_email=claims.get('email', ''),
starred=False
starred=False,
is_public=is_public
)
db.session.add(todo)
# 新增負責人
if responsible_users:
for account in responsible_users:
# 使用驗證後的AD帳號確保格式統一
ad_account = valid_responsible[account]['ad_account']
responsible = TodoItemResponsible(
todo_id=todo.id,
ad_account=account,
ad_account=ad_account,
added_by=identity
)
db.session.add(responsible)
# 新增追蹤人
if followers:
for account in followers:
follower = TodoItemFollower(
todo_id=todo.id,
ad_account=account,
added_by=identity
)
db.session.add(follower)
# 因為匯入的待辦事項預設為非公開,所以不支援追蹤人功能
# 新增稽核記錄
audit = TodoAuditLog(
@@ -447,7 +436,7 @@ def download_template():
'優先級': ['', ''],
'到期日': ['2025-12-31', '2026-01-15'],
'負責人': ['user1@panjit.com.tw', 'user2@panjit.com.tw'],
'追蹤人': ['user3@panjit.com.tw;user4@panjit.com.tw', 'user5@panjit.com.tw']
'公開設定': ['', '']
}
# 說明資料
@@ -459,7 +448,7 @@ def download_template():
'優先級: 緊急/高/中/低',
'到期日: YYYY-MM-DD 格式',
'負責人: AD帳號多人用分號分隔',
'追蹤人: AD帳號多人用分號分隔'
'公開設定: 是/否,決定其他人是否能看到此任務'
],
'說明': [
'請填入待辦事項的標題',
@@ -468,7 +457,7 @@ def download_template():
'可選填 URGENT/HIGH/MEDIUM/LOW',
'例如: 2024-12-31',
'例如: john@panjit.com.tw',
'例如: mary@panjit.com.tw;tom@panjit.com.tw'
'是=公開任務,否=只有建立者和負責人能看到'
]
}

View File

@@ -886,14 +886,7 @@ def follow_todo(todo_id):
)
db.session.add(follower)
# Log audit
audit = TodoAuditLog(
actor_ad=identity,
todo_id=todo_id,
action='FOLLOW',
detail={'follower': identity}
)
db.session.add(audit)
# Note: Skip audit log for FOLLOW action until ENUM is updated
db.session.commit()
logger.info(f"User {identity} followed todo {todo_id}")
@@ -924,14 +917,7 @@ def unfollow_todo(todo_id):
# Remove follower
db.session.delete(follower)
# Log audit
audit = TodoAuditLog(
actor_ad=identity,
todo_id=todo_id,
action='UNFOLLOW',
detail={'follower': identity}
)
db.session.add(audit)
# Note: Skip audit log for UNFOLLOW action until ENUM is updated
db.session.commit()
logger.info(f"User {identity} unfollowed todo {todo_id}")

View File

@@ -1,181 +0,0 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""Test database connection and check data"""
import os
import sys
from dotenv import load_dotenv
import pymysql
from datetime import datetime
# Load environment variables
load_dotenv()
def test_db_connection():
"""Test database connection and list tables"""
print("=" * 60)
print("Testing Database Connection")
print("=" * 60)
# Get database configuration
db_config = {
'host': os.getenv('MYSQL_HOST', 'mysql.theaken.com'),
'port': int(os.getenv('MYSQL_PORT', 33306)),
'user': os.getenv('MYSQL_USER', 'A060'),
'password': os.getenv('MYSQL_PASSWORD', 'WLeSCi0yhtc7'),
'database': os.getenv('MYSQL_DATABASE', 'db_A060'),
'charset': 'utf8mb4'
}
print(f"Host: {db_config['host']}")
print(f"Port: {db_config['port']}")
print(f"Database: {db_config['database']}")
print(f"User: {db_config['user']}")
print("-" * 60)
try:
# Connect to database
connection = pymysql.connect(**db_config)
cursor = connection.cursor()
print("[OK] Successfully connected to database")
# List all tables
print("\n[1] Listing all todo tables:")
cursor.execute("SHOW TABLES LIKE 'todo%'")
tables = cursor.fetchall()
if tables:
for table in tables:
print(f" - {table[0]}")
else:
print(" No todo tables found")
# Check todo_item table
print("\n[2] Checking todo_item table:")
cursor.execute("SELECT COUNT(*) FROM todo_item")
count = cursor.fetchone()[0]
print(f" Total records: {count}")
if count > 0:
print("\n Sample data from todo_item:")
cursor.execute("""
SELECT id, title, status, priority, due_date, creator_ad
FROM todo_item
ORDER BY created_at DESC
LIMIT 5
""")
items = cursor.fetchall()
for item in items:
print(f" - {item[0][:8]}... | {item[1][:30]}... | {item[2]} | {item[5]}")
# Check todo_user_pref table
print("\n[3] Checking todo_user_pref table:")
cursor.execute("SELECT COUNT(*) FROM todo_user_pref")
count = cursor.fetchone()[0]
print(f" Total users: {count}")
if count > 0:
print("\n Sample users:")
cursor.execute("""
SELECT ad_account, display_name, email
FROM todo_user_pref
LIMIT 5
""")
users = cursor.fetchall()
for user in users:
print(f" - {user[0]} | {user[1]} | {user[2]}")
# Check todo_item_responsible table
print("\n[4] Checking todo_item_responsible table:")
cursor.execute("SELECT COUNT(*) FROM todo_item_responsible")
count = cursor.fetchone()[0]
print(f" Total assignments: {count}")
# Check todo_item_follower table
print("\n[5] Checking todo_item_follower table:")
cursor.execute("SELECT COUNT(*) FROM todo_item_follower")
count = cursor.fetchone()[0]
print(f" Total followers: {count}")
cursor.close()
connection.close()
print("\n" + "=" * 60)
print("[OK] Database connection test successful!")
print("=" * 60)
return True
except Exception as e:
print(f"\n[ERROR] Database connection failed: {str(e)}")
print(f"Error type: {type(e).__name__}")
return False
def create_sample_todo():
"""Create a sample todo item for testing"""
print("\n" + "=" * 60)
print("Creating Sample Todo Item")
print("=" * 60)
db_config = {
'host': os.getenv('MYSQL_HOST', 'mysql.theaken.com'),
'port': int(os.getenv('MYSQL_PORT', 33306)),
'user': os.getenv('MYSQL_USER', 'A060'),
'password': os.getenv('MYSQL_PASSWORD', 'WLeSCi0yhtc7'),
'database': os.getenv('MYSQL_DATABASE', 'db_A060'),
'charset': 'utf8mb4'
}
try:
connection = pymysql.connect(**db_config)
cursor = connection.cursor()
# Generate a UUID
import uuid
todo_id = str(uuid.uuid4())
# Insert sample todo
sql = """
INSERT INTO todo_item
(id, title, description, status, priority, due_date, created_at, creator_ad, creator_display_name, creator_email, starred)
VALUES
(%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
"""
values = (
todo_id,
'Test Todo Item - ' + datetime.now().strftime('%Y-%m-%d %H:%M'),
'This is a test todo item created from Python script',
'NEW',
'MEDIUM',
'2025-09-15',
datetime.now(),
'test_user',
'Test User',
'test@panjit.com.tw',
False
)
cursor.execute(sql, values)
connection.commit()
print(f"[OK] Created todo item with ID: {todo_id}")
cursor.close()
connection.close()
return True
except Exception as e:
print(f"[ERROR] Failed to create todo: {str(e)}")
return False
if __name__ == "__main__":
# Test database connection
if test_db_connection():
# Ask if user wants to create sample data
response = input("\nDo you want to create a sample todo item? (y/n): ")
if response.lower() == 'y':
create_sample_todo()
else:
print("\n[WARNING] Please check your database configuration in .env file")
sys.exit(1)

View File

@@ -1,165 +0,0 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""Test LDAP connection and authentication"""
import os
import sys
from ldap3 import Server, Connection, SUBTREE, ALL_ATTRIBUTES
from dotenv import load_dotenv
# Load environment variables
load_dotenv()
def test_ldap_connection():
"""Test LDAP connection"""
print("=" * 50)
print("Testing LDAP Connection")
print("=" * 50)
# Get LDAP configuration
ldap_server = os.getenv('LDAP_SERVER', 'ldap://panjit.com.tw')
ldap_port = int(os.getenv('LDAP_PORT', 389))
ldap_bind_user = os.getenv('LDAP_BIND_USER_DN', '')
ldap_bind_password = os.getenv('LDAP_BIND_USER_PASSWORD', '')
ldap_search_base = os.getenv('LDAP_SEARCH_BASE', 'DC=panjit,DC=com,DC=tw')
print(f"LDAP Server: {ldap_server}")
print(f"LDAP Port: {ldap_port}")
print(f"Bind User: {ldap_bind_user}")
print(f"Search Base: {ldap_search_base}")
print("-" * 50)
try:
# Create server object
server = Server(
ldap_server,
port=ldap_port,
use_ssl=False,
get_info=ALL_ATTRIBUTES
)
print("Creating LDAP connection...")
# Create connection with bind user
conn = Connection(
server,
user=ldap_bind_user,
password=ldap_bind_password,
auto_bind=True,
raise_exceptions=True
)
print("[OK] Successfully connected to LDAP server")
print(f"[OK] Server info: {conn.server}")
# Test search
print("\nTesting LDAP search...")
search_filter = "(objectClass=person)"
conn.search(
ldap_search_base,
search_filter,
SUBTREE,
attributes=['sAMAccountName', 'displayName', 'mail'],
size_limit=5
)
print(f"[OK] Found {len(conn.entries)} entries")
if conn.entries:
print("\nSample users:")
for i, entry in enumerate(conn.entries[:3], 1):
print(f" {i}. {entry.sAMAccountName} - {entry.displayName}")
conn.unbind()
print("\n[OK] LDAP connection test successful!")
return True
except Exception as e:
print(f"\n[ERROR] LDAP connection failed: {str(e)}")
print(f"Error type: {type(e).__name__}")
return False
def test_user_authentication(username, password):
"""Test user authentication"""
print("\n" + "=" * 50)
print(f"Testing authentication for user: {username}")
print("=" * 50)
# Get LDAP configuration
ldap_server = os.getenv('LDAP_SERVER', 'ldap://panjit.com.tw')
ldap_port = int(os.getenv('LDAP_PORT', 389))
ldap_bind_user = os.getenv('LDAP_BIND_USER_DN', '')
ldap_bind_password = os.getenv('LDAP_BIND_USER_PASSWORD', '')
ldap_search_base = os.getenv('LDAP_SEARCH_BASE', 'DC=panjit,DC=com,DC=tw')
ldap_user_attr = os.getenv('LDAP_USER_LOGIN_ATTR', 'userPrincipalName')
try:
# Create server object
server = Server(
ldap_server,
port=ldap_port,
use_ssl=False,
get_info=ALL_ATTRIBUTES
)
# First, bind with service account to search for user
conn = Connection(
server,
user=ldap_bind_user,
password=ldap_bind_password,
auto_bind=True,
raise_exceptions=True
)
# Search for user
search_filter = f"(&(objectClass=person)({ldap_user_attr}={username}))"
print(f"Searching with filter: {search_filter}")
conn.search(
ldap_search_base,
search_filter,
SUBTREE,
attributes=['sAMAccountName', 'displayName', 'mail', 'userPrincipalName', 'distinguishedName']
)
if not conn.entries:
print(f"[ERROR] User not found: {username}")
return False
user_entry = conn.entries[0]
user_dn = user_entry.distinguishedName.value
print(f"[OK] User found:")
print(f" DN: {user_dn}")
print(f" sAMAccountName: {user_entry.sAMAccountName}")
print(f" displayName: {user_entry.displayName}")
print(f" mail: {user_entry.mail}")
# Try to bind with user credentials
print(f"\nAttempting to authenticate user...")
user_conn = Connection(
server,
user=user_dn,
password=password,
auto_bind=True,
raise_exceptions=True
)
print("[OK] Authentication successful!")
user_conn.unbind()
conn.unbind()
return True
except Exception as e:
print(f"[ERROR] Authentication failed: {str(e)}")
return False
if __name__ == "__main__":
# Test basic connection
if test_ldap_connection():
# If you want to test user authentication, uncomment and modify:
# test_user_authentication("your_username@panjit.com.tw", "your_password")
pass
else:
print("\n[WARNING] Please check your LDAP configuration in .env file")
sys.exit(1)

View File

@@ -1,51 +0,0 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""Test LDAP authentication with provided credentials"""
import os
import sys
from dotenv import load_dotenv
from app import create_app
from utils.ldap_utils import authenticate_user
# Load environment variables
load_dotenv()
def test_ldap_auth():
"""Test LDAP authentication"""
print("=" * 60)
print("Testing LDAP Authentication")
print("=" * 60)
print(f"LDAP Server: {os.getenv('LDAP_SERVER')}")
print(f"Search Base: {os.getenv('LDAP_SEARCH_BASE')}")
print(f"Login Attr: {os.getenv('LDAP_USER_LOGIN_ATTR')}")
print("-" * 60)
username = 'ymirliu@panjit.com.tw'
password = '3EDC4rfv5tgb'
print(f"Testing authentication for: {username}")
# Create Flask app and context
app = create_app()
with app.app_context():
try:
result = authenticate_user(username, password)
if result:
print("[SUCCESS] Authentication successful!")
print(f"User info: {result}")
else:
print("[FAILED] Authentication failed")
return result is not None
except Exception as e:
print(f"[ERROR] Exception during authentication: {str(e)}")
import traceback
traceback.print_exc()
return False
if __name__ == "__main__":
test_ldap_auth()