import uuid import hashlib import base64 import logging import random from sqlalchemy import create_engine, Column, Integer, String, MetaData, Table, text from sqlalchemy.orm import sessionmaker from sqlalchemy.ext.declarative import declarative_base # Настройка логирования logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') # Настройка подключения к БД DATABASE_URL = "mssql://mis_aokb_desktop_srv:HLqKhOMwPRjNAVvLIKtc@10.48.89.4:1433/amu_mis_AOKB_prod?driver=ODBC+Driver+18+for+SQL+Server&TrustServerCertificate=yes" # Создаем движок engine = create_engine(DATABASE_URL) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) # Базовый класс для моделей Base = declarative_base() # Модель пользователя class XUser(Base): __tablename__ = 'x_User' UserID = Column(Integer, primary_key=True) GeneralPassword = Column(String(255)) GUID = Column(String(36), nullable=True) # Альтернативный способ через Table (если не хотите использовать ORM) # metadata = MetaData() # x_user_table = Table( # 'x_User', metadata, # Column('UserID', Integer, primary_key=True), # Column('GeneralLogin', String(255)), # Column('GeneralPassword', String(255)), # Column('AuthMode', bool(0)), # Column('FIO', String(255)), # Column('GUID', String(36)) # ) def get_db(): """Получение сессии БД""" db = SessionLocal() try: yield db finally: db.close() def generate_clear_password(length: int = 12) -> str: """ Генерирует случайный пароль без неоднозначных символов Исключает: - Символы, которые можно спутать: 1, l, I, 0, O, o - Специальные символы Args: length: длина пароля Returns: Случайный пароль """ # Безопасные символы: только цифры и буквы (без неоднозначных) safe_digits = '23456789' # без 0,1 safe_uppercase = 'ABCDEFGHJKLMNPQRSTUVWXYZ' # без I, O safe_lowercase = 'abcdefghjkmnpqrstuvwxyz' # без l, o all_chars = safe_digits + safe_uppercase + safe_lowercase # Гарантируем, что пароль содержит хотя бы по одному символу из каждой категории password_chars = [ random.choice(safe_digits), random.choice(safe_uppercase), random.choice(safe_lowercase) ] # Заполняем оставшуюся длину случайными символами из всех категорий for _ in range(length - 3): password_chars.append(random.choice(all_chars)) # Перемешиваем символы random.shuffle(password_chars) return ''.join(password_chars) def compute_hash(password: str, guid: str) -> str: """ Вычисляет хеш пароля по алгоритму из PHP метода Args: password: пароль guid: GUID в строковом формате Returns: base64-encoded хеш """ try: # Проверяем, что GUID валидный uuid_obj = uuid.UUID(guid) except ValueError: logging.warning('Invalid GUID format provided', extra={'guid': guid}) raise ValueError('The provided GUID is not valid') text = guid.upper() bytes_data = (text + password + text).encode('utf-8') # Хеширование SHA1 hash_bytes = hashlib.sha1(bytes_data).digest() # 4-1 итерации повторного хеширования (3 дополнительные итерации) for i in range(3): hash_bytes = hashlib.sha1(hash_bytes).digest() return base64.b64encode(hash_bytes).decode('utf-8') def update_passwords_with_random(user_data_list: list, password_length: int = 12, output_file: str = 'updated_passwords_with_random.txt') -> tuple: """ Обновляет GeneralPassword для списка пользователей с генерацией случайных паролей Args: user_data_list: список кортежей (UserID, GeneralLogin, GUID) password_length: длина генерируемого пароля output_file: имя файла для сохранения результатов Returns: tuple: (количество обновленных пользователей, список результатов) """ results = [] updated_count = 0 # Создаем сессию БД db = SessionLocal() try: with open(output_file, 'w', encoding='utf-8') as f: # Записываем заголовок f.write("UserID\tGeneralLogin\tGUID\tNewPassword\tPasswordHash\tStatus\n") for user_id, login, old_guid in user_data_list: try: # Получаем пользователя из базы данных через ORM user = db.query(XUser).filter(XUser.UserID == user_id).first() if not user: error_msg = f"User with ID {user_id} not found" result_line = f"{user_id}\t{login}\t{old_guid}\t\t\tERROR: User not found" f.write(result_line + "\n") results.append({ 'user_id': user_id, 'login': login, 'guid': old_guid, 'password': '', 'hash': '', 'status': 'ERROR: User not found' }) logging.error(error_msg) continue # Генерируем случайный пароль new_password = generate_clear_password(password_length) # Вычисляем хеш для нового пароля new_hash = compute_hash(new_password, old_guid) # Обновляем пароль пользователя user.GeneralPassword = new_hash # Записываем результат result_line = f"{user_id}\t{login}\t{old_guid}\t{new_password}\t{new_hash}\tSUCCESS" f.write(result_line + "\n") results.append({ 'user_id': user_id, 'login': login, 'guid': old_guid, 'password': new_password, 'hash': new_hash, 'status': 'SUCCESS' }) updated_count += 1 logging.info(f'Password updated for user ID: {user_id}, Login: {login}') except Exception as e: error_msg = f"Error updating user {user_id}: {str(e)}" result_line = f"{user_id}\t{login}\t{old_guid}\t\t\tERROR: {str(e)}" f.write(result_line + "\n") results.append({ 'user_id': user_id, 'login': login, 'guid': old_guid, 'password': '', 'hash': '', 'status': f'ERROR: {str(e)}' }) logging.error(error_msg) # Коммитим все изменения в БД db.commit() logging.info(f'Successfully updated passwords for {updated_count} users') logging.info(f'Results saved to {output_file}') return updated_count, results except Exception as e: # Откатываем изменения в случае ошибки db.rollback() logging.error(f'Error processing users: {str(e)}') raise finally: # Закрываем сессию db.close() # Альтернативная версия с использованием прямого SQL (без ORM) # def update_passwords_direct_sql(user_data_list: list, password_length: int = 12, # output_file: str = 'updated_passwords_direct.txt') -> tuple: # """ # Обновляет пароли используя прямое SQL выполнение # """ # results = [] # updated_count = 0 # # Создаем сессию БД # db = SessionLocal() # try: # with open(output_file, 'w', encoding='utf-8') as f: # f.write("UserID\tGeneralLogin\tGUID\tNewPassword\tPasswordHash\tStatus\n") # for user_id, login, old_guid in user_data_list: # try: # # Генерируем случайный пароль # new_password = generate_clear_password(password_length) # # Вычисляем хеш # new_hash = compute_hash(new_password, old_guid) # # Прямой SQL запрос для обновления # stmt = x_user_table.update().where( # x_user_table.c.UserID == user_id # ).values( # GeneralPassword=new_hash # ) # # Выполняем запрос # result = db.execute(stmt) # if result.rowcount == 0: # # Пользователь не найден # result_line = f"{user_id}\t{login}\t{old_guid}\t\t\tERROR: User not found" # results.append({ # 'user_id': user_id, 'login': login, 'guid': old_guid, # 'password': '', 'hash': '', 'status': 'ERROR: User not found' # }) # else: # # Успешное обновление # result_line = f"{user_id}\t{login}\t{old_guid}\t{new_password}\t{new_hash}\tSUCCESS" # results.append({ # 'user_id': user_id, 'login': login, 'guid': old_guid, # 'password': new_password, 'hash': new_hash, 'status': 'SUCCESS' # }) # updated_count += 1 # f.write(result_line + "\n") # logging.info(f'Processed user ID: {user_id}, Login: {login}') # except Exception as e: # error_msg = f"Error updating user {user_id}: {str(e)}" # result_line = f"{user_id}\t{login}\t{old_guid}\t\t\tERROR: {str(e)}" # f.write(result_line + "\n") # results.append({ # 'user_id': user_id, 'login': login, 'guid': old_guid, # 'password': '', 'hash': '', 'status': f'ERROR: {str(e)}' # }) # logging.error(error_msg) # # Коммитим изменения # db.commit() # logging.info(f'Successfully updated passwords for {updated_count} users') # return updated_count, results # except Exception as e: # db.rollback() # logging.error(f'Error processing users: {str(e)}') # raise # finally: # db.close() def create_user_list_from_data() -> list: """ Создает список пользователей из предоставленных данных """ user_data = [ (813, 'ЦыпленковаНИ', '9D29B54E-6896-4D91-BE80-B337874768E3'), (814, 'МакушеваЕА', '89827204-8CD2-40A9-A9EA-1A31B4B4748B'), (817, 'КравченкоЕО', '0DD1FCD7-7F16-4EA4-848A-E6B7D7D59F9B'), (818, 'ИвановаИН', '19B968CA-94D7-4747-8903-B6EE80CD8694'), (838, 'ПехЕЮ', '1C3BC2A1-6827-414E-9E4E-626F4AD212CF'), (1348, 'БрагинаАН', 'D24F0EF4-13C4-4BC9-BA46-ED8B3385EB33'), (1366, 'ТимченкоНИ', '22724FC1-79BD-46F8-9368-2B51905B45D1'), (1871, 'БурчакГА', '39FA64EB-AFD1-4476-B552-2C4EFBEE64D9'), (1875, 'ДымченкоВЛ', '8678D82B-43E2-4563-A5A3-AD0D946C2752'), (1876, 'ЁминаНВ', 'FF563C99-095D-43F6-900A-ECBAB729A424'), (1882, 'ПетяеваИА', 'A187CBDF-D7E0-41D6-933F-C2D3B6076329'), (1884, 'СивухинаИА', '0C0B838E-CA04-422A-9CE0-06F8AE57BFCE'), (1958, 'ЛандышеваМВ', '94283067-72C4-4F92-955E-7DCD1C5B1DA4'), (1967, 'ПоповичЗН', 'E317397F-D765-42A5-A792-535108359083'), (1977, 'ВафинаДМ', 'F2277378-504E-4E7E-9786-1400D442E430'), (1988, 'СтрелковаЛВ', 'CF5C742B-CB93-42F9-8376-2074DA2D97B2'), (2022, 'АлешковичЛЕ', '9B2473DE-25BD-42BA-BC66-17F85A5D4A48'), (2035, 'ЮСН', '8C22B66C-D74E-43BC-AC71-571102D686BF'), (2038, 'КрамаренкоСА', '5205F9BF-0A4E-45C2-9BF5-7F851D7A848F'), (2108, 'КравцоваДМ', 'F642B77B-1BC2-4173-955A-1A41A83868AD'), (2132, 'ПушенковаОА', 'F9EBE958-217A-4415-93D1-5299A11B97F7'), (2221, 'ТихоноваТЮ', 'C673B1AC-D22F-428E-BF11-84F078F18073'), (2253, 'ЮрковаАЕ', '662C9467-9325-4666-9847-CC2F5FE2B329'), (2331, 'ОкулОЮ', 'B3A9B440-B94E-4B73-9F07-D4F1BE4C9DA3'), (2336, 'МардановаГЭ', '3248B6C4-685D-4ABC-BF4D-2F94AF47A30C'), (2353, 'ГуковскаяЕВ', '23271681-9331-4192-A674-BD8FDF5C5EBC'), (2354, 'ГуроваЕЮ', 'AE4EE178-B5BB-4B8E-8D0B-56695D0A29A0'), (2370, 'ПелеховаЕИ', '2D1C4169-4268-4093-A380-8A3EAB7B93A3'), (2404, 'ДорожинскаяНВ', '2E182439-4C54-45AF-B74B-D5C5228E1FA3'), (2451, 'ПолонецВН', 'CE045B5E-EE2C-49FD-8418-8A896EB22D7B'), (2484, 'КоростелёваТН', 'FAC707CC-6FE3-4F5E-A192-9A311542371E'), (2562, 'ЛиМР', '8DA63EDD-1425-485D-B505-8E9D47D8703C'), (2592, 'ЧугуноваОБ', '0F58B793-025E-4878-BDF5-B90F054A616C'), (2595, 'МакушкинаКН', '05AA98EB-A26D-485E-8479-0BFACED5511D'), (2606, 'ГоЛВ', '0861CA27-52DD-4C04-A696-69F6972959C9'), (2687, 'ЗубареваСВ', 'FA45DABC-B077-4AC8-9A23-48F66C1C1503'), (2710, 'МиловидоваМА', '7995CC03-7E50-495B-8122-4FB62416BA2C'), (2735, 'ДудкаДА', '6EF7DFD3-904D-4E97-81BC-5164654B20D9'), (2738, 'ГоринаКА', '450FF465-82DE-4031-9645-5DA4EF4CC59E'), (2746, 'ЗайнутдиноваНМ', '13D01C4C-F2E3-47A8-8CD3-D770C091983F'), (2764, 'МакаренкоЕА', '87A80BF1-D8EF-479C-8BCC-66B002F02733'), (2811, 'БуторинаАЮ', '92020D1B-4CBE-4A11-AE50-5A70E9EAC004'), (2814, 'БеловаИВ', '8035FD51-00F0-477A-AE24-DB72A4399B27'), (2815, 'КоровинаЛВ', '4FB57E8D-EF2D-4769-960F-6DC1740FF37E'), (2828, 'ФроловаОА', 'A47CB163-1896-49E1-BC2A-165B257FF9AE'), (2832, 'ЛуговскаяВА', 'A6E22703-E966-46BC-A5AA-753DA8061B39'), (3851, 'ВологжинаВЗ', '24ED005C-D7CF-4524-849C-1ED014DAD538'), (3879, 'БурковаЛВ', '18DD1B9B-E37C-448F-BD01-722E45E15F4C') ] return user_data def test_database_connection(): """Тестирует подключение к БД""" try: db = SessionLocal() # Простой запрос для проверки подключения result = db.execute(text("SELECT 1")) db.close() print("✓ Подключение к БД успешно") return True except Exception as e: print(f"✗ Ошибка подключения к БД: {e}") return False def test_password_generation(): """Тестирует генерацию паролей""" print("Тест генерации паролей:") for i in range(5): password = generate_clear_password(12) print(f" Пароль {i+1}: {password}") def main(): """Основная функция для выполнения обновления паролей""" print("Начало обновления паролей...") # Тестируем подключение к БД if not test_database_connection(): return # Тестируем генерацию паролей test_password_generation() # Получаем список пользователей user_list = create_user_list_from_data() print(f"Найдено пользователей для обработки: {len(user_list)}") # Выберите метод обновления: # 1. Через ORM (рекомендуется) print("\nИспользование ORM метода...") try: updated_count, results = update_passwords_with_random( user_list, password_length=12, output_file='updated_passwords_orm.txt' ) print(f"Результаты (ORM):") print(f" Успешно обновлено: {updated_count} пользователей") print(f" Ошибки: {len(user_list) - updated_count} пользователей") except Exception as e: print(f"Ошибка при обновлении паролей (ORM): {e}") # 2. Через прямое SQL (альтернатива) # print("\nИспользование прямого SQL метода...") # try: # updated_count, results = update_passwords_direct_sql( # user_list, # password_length=12, # output_file='updated_passwords_sql.txt' # ) # print(f"Результаты (SQL):") # print(f" Успешно обновлено: {updated_count} пользователей") # print(f" Ошибки: {len(user_list) - updated_count} пользователей") # except Exception as e: # print(f"Ошибка при обновлении паролей (SQL): {e}") # Создаем удобный отчет if 'results' in locals(): create_password_report(results) def create_password_report(results: list, filename: str = 'password_report.txt'): """Создает удобный отчет с паролями""" with open(filename, 'w', encoding='utf-8') as f: f.write("ОТЧЕТ ОБ ОБНОВЛЕНИИ ПАРОЛЕЙ\n") f.write("=" * 50 + "\n\n") success_results = [r for r in results if r['status'] == 'SUCCESS'] error_results = [r for r in results if r['status'] != 'SUCCESS'] f.write(f"Всего пользователей: {len(results)}\n") f.write(f"Успешно обновлено: {len(success_results)}\n") f.write(f"Ошибок: {len(error_results)}\n\n") f.write("СПИСОК ПАРОЛЕЙ:\n") f.write("-" * 50 + "\n") for result in success_results: f.write(f"Логин: {result['login']}\n") f.write(f"UserID: {result['user_id']}\n") f.write(f"Пароль: {result['password']}\n") f.write(f"GUID: {result['guid']}\n") f.write(f"Хеш: {result['hash']}\n") f.write("-" * 30 + "\n") if error_results: f.write("\nОШИБКИ:\n") f.write("-" * 30 + "\n") for error in error_results: f.write(f"UserID: {error['user_id']}, Логин: {error['login']}\n") f.write(f"Ошибка: {error['status']}\n") f.write("-" * 20 + "\n") if __name__ == "__main__": main()