Text Share Online

from ldap3 import Server, Connection, SAFE_SYNC, SUBTREE, MODIFY_REPLACE , MODIFY_ADD , SUBTREE
from ldap3.core.exceptions import LDAPException
from ldap3.extend.microsoft.modifyPassword import ad_modify_password
from datetime import datetime ,timezone
#from pypyodbc import output
import os ,sys
import cx_Oracle
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), ‘..’,’..’,’..’)))
from RAM import settings
from iManage_logs import update_log
import json
import django
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
PROJECT_DIR = os.path.join(BASE_DIR, ‘/app/RAM’) #uat
sys.path.append(PROJECT_DIR)
os.environ.setdefault(‘DJANGO_SETTINGS_MODULE’, ‘RAM.settings’)
django.setup()
import generic_gapi_email_sender as gapi
from django.template.loader import render_to_string
from MASTER.models import *
from MASTER.views import update_is_active
import secrets
import string
import base64
import secrets

# You can tune this length to your policy (e.g., 16, 20, 24)
PASSWORD_LENGTH = 16

# Exclude characters that are commonly problematic in some contexts.
# Since unicodePwd is quoted and encoded UTF-16-LE, most characters are safe.
# However, we keep the set clean and policy-compliant.
_SPECIALS = “!@#$%^&*-_=+;:,.?/”

def generate_strong_password(length: int = PASSWORD_LENGTH) -> str:
# Enforce a reasonable minimum
length = max(12, int(length))

alphabet_lower = string.ascii_lowercase
alphabet_upper = string.ascii_uppercase
digits = string.digits
specials = _SPECIALS

# Decide how many digits: 3â€“6
num_digits = secrets.choice(range(3, 7))

# Mandatory components
one_special = [secrets.choice(specials)]
one_lower = [secrets.choice(alphabet_lower)]
one_upper = [secrets.choice(alphabet_upper)]
digit_block = [secrets.choice(digits) for _ in range(num_digits)]

# Remaining characters should be letters only (no extra symbols)
letters_pool = alphabet_lower + alphabet_upper
remaining_len = length – (1 + 1 + 1 + num_digits) # special + lower + upper + digits
remaining_letters = [secrets.choice(letters_pool) for _ in range(remaining_len)]

# Assemble and shuffle
pwd_list = one_special + one_lower + one_upper + digit_block + remaining_letters
secrets.SystemRandom().shuffle(pwd_list)
return ”.join(pwd_list)

def generate_time_stamp():
time1 = datetime.now()
return time1

# === CONFIGURATION ===
LDAP_HOST = settings.LLDAP_HOST
PORT = settings.LDAP_PORT
LDAP_USER = settings.LLDAP_USER
LDAP_PASSWORD = settings.LLDAP_PASSWORD
BASE_DN = ‘DC=ratnakarbank,DC=in’
BOT_ID_OU =’OU=RBL Temp Computers,DC=ratnakarbank,DC=in’
GIFT_ID_OU = ‘OU=Misc,OU=RatnakarBank,DC=ratnakarbank,DC=in’
GENERIC_ID_OU=’OU=Misc_win10,OU=Misc,OU=RatnakarBank,DC=ratnakarbank,DC=in’

def convert_to_filetime(dt):
if dt.tzinfo is None:
dt = dt.replace(tzinfo=timezone.utc)
else:
dt = dt.astimezone(timezone.utc)

filetime = int((dt – datetime(1601, 1, 1, tzinfo=timezone.utc)).total_seconds() * 10**7)
return str(filetime)

def ldap_connection():
try:
server = Server(LDAP_HOST, PORT, use_ssl=True)
conn = Connection(server, user=LDAP_USER, password=LDAP_PASSWORD, client_strategy=SAFE_SYNC, auto_bind=True)
# print(“ldap connected”)
return conn
except LDAPException as e:
# print(f”[ERROR] LDAP connection failed: {e}”)
update_log(req=”, res=e, err_type=’T’, error_code=”,error_desc=f”[ERROR] LDAP connection failed: {e}”, error_source=’GID_BID_LDAP_ALERT’,res_time=datetime.now(), tp_req=”, service_name=”, file_name=”,cron_stime=”, sh_name=’GID_BID_BACKEND.sh’)
return None

conn = ldap_connection()

def fetch_latest_id(OU):
success = conn.search(
search_base=OU,
search_filter='(objectClass=user)’,
search_scope=SUBTREE,
attributes=[‘sAMAccountName’, ‘whenCreated’] )

users = [
(entry[‘attributes’][‘sAMAccountName’], entry[‘attributes’][‘whenCreated’])
for entry in success[2]
if ‘attributes’ in entry and ‘sAMAccountName’ in entry[‘attributes’]
]

users_sorted = sorted(users, key=lambda x: x[1], reverse=True)

if users_sorted:
latest_user = users_sorted[0][0]
return latest_user
else:
return False

def increment_samaccountname(sam):
prefix = ”.join(filter(str.isalpha, sam)) # Extract letters (e.g., ‘B’)
number = ”.join(filter(str.isdigit, sam)) # Extract digits (e.g., ‘001’)
new_number = str(int(number) + 1).zfill(len(number)) # Increment and pad with zeros
return prefix + new_number

def fetch_dn(mail,IM_MODULE_NAME):
search_filter = f”(userPrincipalName={mail})”
print(‘search_filter..’, search_filter)
success = conn.search(search_base=BASE_DN,search_filter= search_filter,search_scope=SUBTREE,attributes=[‘cn’, ‘sn’, ‘name’, ’employeeID’, ‘mail’, ‘memberof’, ‘sAMAccountName’,’manager’])
print(‘success..’, success)
if ‘attributes’ in success[2][0]:
manager_dn = ”
if IM_MODULE_NAME in [“Generic Id With Email (Interactive)”,”Generic Id Without Email (Interactive)”,”Generic Id With Email (Non-Interactive)”,”Generic Id Without Email (Non-Interactive)”]:
manager_dn = success[2][0][‘dn’]
else:
manager_dn = success[2][0][‘attributes’][‘manager’]
return manager_dn
else:
return False

def fetch_user_dn(emp_id):
search_filter = f”(&(sAMAccountName={emp_id})(objectclass=person))”
success = conn.search(‘DC=ratnakarbank,DC=in’,search_filter,attributes=[‘cn’, ‘sn’, ‘name’, ’employeeID’, ‘mail’, ‘memberof’, ‘sAMAccountName’])
if success and len(success) > 2 and success[2]:
user_dn = success[2][0][‘dn’] # This is the user’s own DN
# print(“User DN:”, user_dn)
return user_dn
else:
return False

def build_email(first_name, last_name):
# Normalize to lower and basic sanitization; adapt if you need stricter rules
def normalize(s):
return ”.join(ch for ch in s.lower() if ch.isalnum() or ch in [‘.’, ‘-‘, ‘_’])
if last_name:
local = f”{normalize(first_name)}.{normalize(last_name)}”
else:
local = normalize(first_name)
domain = “rblbank.com”
return f”{local}@{domain}”

def generic_bot_gift_id_creation(B_G_ID_NAME, DEPARTMENT, CONTACT_NO, OWNER_OF_GID_BID,TYPE_OF_GID_BID, TEMPORARY_OR_PERMANENT, IF_TEMP_DATE_OF_ID_DELETED,HOSTNAME, ID_TYPE, IM_MODULE_NAME, REQUEST_FOR_ID,manager_name_for_dn):
name_parts = B_G_ID_NAME.split(‘ ‘)
response = {}
user_attributes = {}
manager_dn = fetch_dn(OWNER_OF_GID_BID,IM_MODULE_NAME)
response[‘password’] = ”
if IM_MODULE_NAME in [“Generic Id With Email (Interactive)”,”Generic Id Without Email (Interactive)”,”Generic Id With Email (Non-Interactive)”,”Generic Id Without Email (Non-Interactive)”]:
gid_bid_desc_for_id_creation = ‘Owner :’ + OWNER_OF_GID_BID.split(‘.’)[0] + ” ” + OWNER_OF_GID_BID.split(‘.’)[1].split(‘@’)[0]
else:
gid_bid_desc_for_id_creation = f’Owner :{manager_name_for_dn}’

if manager_dn == False:
error_message = f’Manager DN is null for {ID_TYPE} ID creation’
res_datetime = datetime.now()
response[‘status’] = False
response[‘response’] = error_message
update_log(
req=”, res=error_message, err_type=’T’, error_code=”,
error_desc=str(error_message), error_source=f'{ID_TYPE} ID CREATION’,
res_time=res_datetime, tp_req=”, service_name=”, file_name=”,
cron_stime=”, sh_name=’AD_ID_CREATION.sh’
)
return response

if TEMPORARY_OR_PERMANENT == ‘Temporary’:
expiry = convert_to_filetime(IF_TEMP_DATE_OF_ID_DELETED)
user_attributes[‘accountExpires’] = expiry

sAMAccountName = ”
# Determine OU and fetch latest ID based on ID_TYPE
if ID_TYPE == ‘BOT’:
latest_id = fetch_latest_id(BOT_ID_OU)
user_dn_ou = “OU=RBL Temp Computers,DC=ratnakarbank,DC=in”
sAMAccountName = increment_samaccountname(latest_id)
elif ID_TYPE == ‘GIFT’:
latest_id = ‘GIFT’ + str(REQUEST_FOR_ID)
user_dn_ou = “OU=Misc,OU=RatnakarBank,DC=ratnakarbank,DC=in”
sAMAccountName = latest_id
elif ID_TYPE == ‘GENERIC’:
latest_id = fetch_latest_id(GENERIC_ID_OU)
user_dn_ou = “OU=Misc_win10,OU=Misc,OU=RatnakarBank,DC=ratnakarbank,DC=in”
sAMAccountName = increment_samaccountname(latest_id)
else:
response[‘status’] = False
response[‘response’] = ‘Invalid ID_TYPE specified’
return response

if latest_id == False:
error_message = f’Unable to fetch latest {ID_TYPE} ID’
res_datetime = datetime.now()
response[‘status’] = False
response[‘response’] = error_message
update_log(
req=”, res=error_message, err_type=’T’, error_code=”,
error_desc=str(error_message), error_source=f'{ID_TYPE} ID CREATION’,
res_time=res_datetime, tp_req=”, service_name=”, file_name=”,
cron_stime=”, sh_name=’AD_ID_CREATION.sh’
)
return response

first_name = name_parts[0]
last_name = ”.join(name_parts[1:]) if len(name_parts) > 1 else ”
full_name = f”{first_name} {last_name}”.strip()

primary_email = build_email(first_name, last_name)
if primary_email == False:
error_message = f’Unable to build email of {first_name} [EMP ID:{B_G_ID_NAME}]’
res_datetime = datetime.now()
response[‘status’] = False
response[‘response’] = error_message
update_log(
req=”, res=error_message, err_type=’T’, error_code=”,
error_desc=str(error_message), error_source=f'{ID_TYPE} ID CREATION’,
res_time=res_datetime, tp_req=”, service_name=”, file_name=”,
cron_stime=”, sh_name=’AD_ID_CREATION.sh’
)
return response

# — add exchange attributes only if IM_MODULE_NAME does NOT contain ‘without’ and bot and gift id —
skip_keywords = (‘without’, ‘bot’, ‘gift’)

should_add_exchange_attrs = not any(
kw in IM_MODULE_NAME.lower() for kw in skip_keywords
)

proxy_addresses = []
if should_add_exchange_attrs:
# Primary proxy address
proxy_addresses = [f”SMTP:{primary_email}”]
alias_simple = f”{sAMAccountName}” + “@rbldwhoutlook.mail.onmicrosoft.com”
proxy_addresses.append(f”smtp:{alias_simple}”)
# Using primary email as the target (adjust if your routing requires different domain)
user_attributes[‘targetAddress’] = f”smtp:{alias_simple}”

createdUserPrincipalName = ”
if IM_MODULE_NAME in (‘GIFT Id (Interactive)’):
createdUserPrincipalName = f”{latest_id}@rblbank.com”
elif last_name and IM_MODULE_NAME not in (‘GIFT Id (Interactive)’):
createdUserPrincipalName =f”{first_name}.{last_name}@rblbank.com”
user_attributes[‘Mobile’] = CONTACT_NO
else:
createdUserPrincipalName = f”{first_name}@rblbank.com”
user_attributes[‘Mobile’] = CONTACT_NO
# Base user attributes
user_attributes.update({
‘objectClass’: [‘top’, ‘person’, ‘organizationalPerson’, ‘user’],
‘cn’: full_name,
‘sn’:last_name if last_name and last_name.strip() else first_name,
‘givenName’: first_name,
‘displayName’: full_name,
‘name’: full_name,
‘userPrincipalName’: createdUserPrincipalName,
‘sAMAccountName’: sAMAccountName,
‘manager’: manager_dn,
‘Department’: DEPARTMENT,
‘Company’: ‘RBL BANK LTD’,
‘description’: gid_bid_desc_for_id_creation,
})

# Only include proxyAddresses if we decided to add them
if should_add_exchange_attrs and proxy_addresses:
user_attributes[‘mail’] = primary_email
user_attributes[‘proxyAddresses’] = proxy_addresses

user_dn = f”CN={full_name},{user_dn_ou}”
new_password = generate_strong_password()
result = ad_user_creation(user_attributes, user_dn, IM_MODULE_NAME, new_password)
response[‘status’] = result[‘status’]
response[‘password’] = new_password
response[‘response’] = sAMAccountName if result[‘status’] else result[‘message’]
# print(user_attributes)
return response

def ad_user_creation(user_attributes, user_dn, IM_MODULE_NAME, new_password):
print(‘user dn..’, user_dn)
creation_res = {}
user_added = False
group_dns = [
‘CN=self_help_all_users,OU=Security Groups,OU=Groups,DC=ratnakarbank,DC=in’,
‘CN=Zscaler_Proxy_Users_Policy_Group,OU=Security Groups,OU=Groups,DC=ratnakarbank,DC=in’,
‘CN=emcscheduledarchiveactivity,OU=Distribution Groups,OU=Groups,DC=ratnakarbank,DC=in’
]
# Step 1: Try to add the user
print(‘hereeeeeee’)
result = conn.add(user_dn, attributes=user_attributes)
print(’00-~~~~~~~~~~~~~~~~-‘,result)
if not result[0]:
error = conn.result
if error[‘description’] == ‘entryAlreadyExists’:
# print(“User already exists.”)
creation_res[‘status’] = True
creation_res[‘response’] = “User already exists”
creation_res[‘message’] = “User already exists”
return creation_res
else:
creation_res[‘status’] = False
creation_res[‘response’] = error[‘description’]
creation_res[‘message’] = f”Unable to add user – {result[1][‘message’]}”
# print(‘Failed at user creation’)
return creation_res
user_added = True

# Step 2: Add to group if applicable
if IM_MODULE_NAME in (‘Generic Id With Email (Interactive)’,’Generic Id With Email (Non-Interactive)’):
group_dns.append(‘CN=E1-With-MailAccess-Only,OU=Security Groups,OU=Groups,DC=ratnakarbank,DC=in’)
group_dns.append(‘CN=PreventOutboundEmails-All Domain,OU=Security Groups,OU=Groups,DC=ratnakarbank,DC=in’)

for group_dn in group_dns:
result_modify = conn.modify(group_dn, {‘member’: [(MODIFY_ADD, [user_dn])]})
print(’01-~~~~~~~~~~~~~~~~-‘,result_modify)
if not result_modify[0]:
# ROLLBACK: delete the created user
if user_added:
try:
conn.delete(user_dn)
except Exception as e:
# print(f’Unable to delete created user as rollback – {e}’)
update_log(req=”, res=f’Unable to delete created user as rollback – {e}’, err_type=’T’, error_code=”,error_desc=f’Unable to delete created user as rollback – {e}’, error_source=’GID_BID_DELETE_ROLLBACK_ALERT’,res_time=datetime.now(), tp_req=”, service_name=”, file_name=”,cron_stime=”, sh_name=’GID_BID_BACKEND.sh’)
error = conn.result
creation_res[‘status’] = False
creation_res[‘response’] = error[‘description’]
creation_res[‘message’] = f”Unable to add groups – {result_modify[1][‘message’]}”
# print(‘Failed at group addition’)
return creation_res

# Step 3: Set password
password_value = f'”{new_password}”‘.encode(‘utf-16-le’)
pass_res= conn.modify(user_dn, {
‘unicodePwd’: [(MODIFY_REPLACE, [password_value])]
})
# print(‘1-~~~~~~~~~~~~~~~~-‘,pass_res)

if not pass_res[0]:
# ROLLBACK: delete the created user
if user_added:
try:
conn.delete(user_dn)
except Exception as _e:
# print(f’Unable to delete created user as rollback – {e}’)
update_log(req=”, res=f’Unable to delete created user as rollback – {e}’, err_type=’T’, error_code=”,error_desc=f’Unable to delete created user as rollback – {e}’, error_source=’GID_BID_DELETE_ROLLBACK_ALERT’,res_time=datetime.now(), tp_req=”, service_name=”, file_name=”,cron_stime=”, sh_name=’GID_BID_BACKEND.sh’)
error = conn.result
creation_res[‘status’] = False
creation_res[‘response’] = error[‘description’]
creation_res[‘message’] = f”Unable to set password – {pass_res[1][‘message’]}”
# print(“Failed to update password”)
return creation_res

# Step 4: Enable account
pass_modify = conn.modify(user_dn, {‘userAccountControl’: [(MODIFY_REPLACE, [512])]})
# print(‘2-~~~~~~~~~~~~~~~~-‘,pass_modify)
if not pass_modify[0]:
# ROLLBACK: delete the created user
if user_added:
try:
conn.delete(user_dn)
except Exception as _e:
# print(f’Unable to delete created user as rollback – {e}’)
update_log(req=”, res=f’Unable to delete created user as rollback – {e}’, err_type=’T’, error_code=”,error_desc=f’Unable to delete created user as rollback – {e}’, error_source=’GID_BID_DELETE_ROLLBACK_ALERT’,res_time=datetime.now(), tp_req=”, service_name=”, file_name=”,cron_stime=”, sh_name=’GID_BID_BACKEND.sh’)
error = conn.result
creation_res[‘status’] = False
creation_res[‘response’] = error[‘description’]
creation_res[‘message’] = f”Unable to enable account- {pass_modify[1][‘message’]}”
# print(“Failed to enable account”)
return creation_res

# Step 5: Force password change at next login
pass_enf = conn.modify(user_dn, {‘pwdLastSet’: [(MODIFY_REPLACE, [0])]})
# print(‘3-~~~~~~~~~~~~~~~~-‘,pass_enf)
if not pass_enf[0]:
# ROLLBACK: delete the created user
if user_added:
try:
conn.delete(user_dn)
except Exception as _e:
# print(f’Unable to delete created user as rollback – {e}’)
update_log(req=”, res=f’Unable to delete created user as rollback – {e}’, err_type=’T’, error_code=”,error_desc=f’Unable to delete created user as rollback – {e}’, error_source=’GID_BID_DELETE_ROLLBACK_ALERT’,res_time=datetime.now(), tp_req=”, service_name=”, file_name=”,cron_stime=”, sh_name=’GID_BID_BACKEND.sh’)
error = conn.result
creation_res[‘status’] = False
creation_res[‘response’] = error[‘description’]
creation_res[‘message’] = f”Unable to add force password change at next login – {pass_enf[1][‘message’]}”
# print(“Failed to enforce password change”)
return creation_res

# Step 6: Adding hostname to user account if non-interactive need to add Non-Interactive string.
if any(x in IM_MODULE_NAME.lower() for x in (‘generic’, ‘bot’)):
host_add = conn.modify(user_dn, {‘userWorkstations’: [(MODIFY_REPLACE, [HOSTNAME])]})
# print(‘4-~~~~~~~~~~~~~~~~-‘,host_add)
if not host_add[0]:
# ROLLBACK: delete the created user
if user_added:
try:
conn.delete(user_dn)
except Exception as _e:
# print(f’Unable to delete created user as rollback – {e}’)
update_log(req=”, res=f’Unable to delete created user as rollback – {e}’, err_type=’T’, error_code=”,error_desc=f’Unable to delete created user as rollback – {e}’, error_source=’GID_BID_DELETE_ROLLBACK_ALERT’,res_time=datetime.now(), tp_req=”, service_name=”, file_name=”,cron_stime=”, sh_name=’GID_BID_BACKEND.sh’)
error = conn.result
creation_res[‘status’] = False
creation_res[‘response’] = error[‘description’]
creation_res[‘message’] = f”Unable to add hostname to user account – {host_add[1][‘message’]}”
# print(“Failed to Adding hostname”)
return creation_res

# Success
creation_res[‘status’] = True
creation_res[‘response’] = “User created and configured successfully”
creation_res[‘message’] = f”created and configured successfully.”
return creation_res

def reset_ad_password(user_principal_name):
try:
search_filter = f”(&(sAMAccountName={user_principal_name})(objectclass=person))”
record = conn.search(
‘DC=ratnakarbank,DC=in’,
search_filter,
attributes=[‘cn’, ‘sn’, ‘name’, ’employeeID’, ‘mail’, ‘memberof’, ‘sAMAccountName’]
)

if not record:
return {‘status’: False, ‘response’: f”User {user_principal_name} not found in AD”}

user_dn = record[2][0][‘dn’]

# Reset password using AD extension
new_password = generate_strong_password()
result = ad_modify_password(conn, user_dn, new_password, old_password=None)

if result:
return {‘status’: True, ‘response’: f”Password for {user_principal_name} reset successfully.”, ‘password’: new_password}
else:
return {‘status’: False, ‘response’: f”Failed to reset password for {user_principal_name}: {conn.result}”}

except Exception as e:
return {‘status’: False, ‘response’: str(e)}

def change_ad_owner(user_principal_name, new_owner_dn, manager_name):
try:
search_filter = f”(&(sAMAccountName={user_principal_name})(objectclass=person))”
record = conn.search(
‘DC=ratnakarbank,DC=in’,
search_filter,
attributes=[‘cn’, ‘sn’, ‘name’, ’employeeID’, ‘mail’, ‘memberof’, ‘sAMAccountName’]
)

if not record:
return {‘status’: False, ‘response’: f”User {user_principal_name} not found in AD”}

user_dn = record[2][0][‘dn’]

# Change owner by updating ‘managedBy’ attribute
result = conn.modify(user_dn, {‘manager’: [(MODIFY_REPLACE, [new_owner_dn])],’description’: [(MODIFY_REPLACE, [f”Owner : {manager_name}”])]})

if result:
return {‘status’: True, ‘response’: f”Owner for {user_principal_name} changed successfully”}
else:
return {‘status’: False, ‘response’: f”Failed to change owner for {user_principal_name}: {conn.result}”}

except Exception as e:
return {‘status’: False, ‘response’: str(e)}

def update_ad_mobile_number(user_principal_name, new_mobile_number):
try:
search_filter = f”(&(sAMAccountName={user_principal_name})(objectclass=person))”
record = conn.search(
‘DC=ratnakarbank,DC=in’,
search_filter,
attributes=[‘cn’, ‘sn’, ‘name’, ’employeeID’, ‘mail’, ‘memberof’, ‘sAMAccountName’]
)

if not record:
return {‘status’: False, ‘response’: f”User {user_principal_name} not found in AD”}

user_dn = record[2][0][‘dn’]

# Modify the mobile number
result = conn.modify(user_dn, {‘Mobile’: [(MODIFY_REPLACE, [new_mobile_number])]})

if result:
return {‘status’: True, ‘response’: f”Mobile number for {user_principal_name} updated successfully.”}
else:
return {‘status’: False, ‘response’: f”Failed to update mobile number: {conn.result}”}

except Exception as e:
return {‘status’: False, ‘response’: str(e)}

def update_id_types(user_principal_name,new_hostname):
try:
search_filter = f”(&(sAMAccountName={user_principal_name})(objectclass=person))”
record = conn.search(
‘DC=ratnakarbank,DC=in’,
search_filter,
attributes=[‘cn’, ‘sn’, ‘name’, ’employeeID’, ‘mail’, ‘memberof’, ‘sAMAccountName’]
)

if not record:
return {‘status’: False, ‘response’: f”User {user_principal_name} not found in AD”}

user_dn = record[2][0][‘dn’]

result = conn.modify(user_dn, {‘userWorkstations’: [(MODIFY_REPLACE, [new_hostname])]})

if result:
return {‘status’: True, ‘response’: f”userWorkstations for {user_principal_name} updated successfully.”}
else:
return {‘status’: False, ‘response’: f”Failed to update userWorkstations: {conn.result}”}

except Exception as e:
return {‘status’: False, ‘response’: str(e)}

def update_ad_account_expiry(user_principal_name, expiry_date):
try:
filetime = convert_to_filetime(expiry_date)

search_filter = f”(&(sAMAccountName={user_principal_name})(objectclass=person))”
record = conn.search(
‘DC=ratnakarbank,DC=in’,
search_filter,
attributes=[‘cn’, ‘sn’, ‘name’, ’employeeID’, ‘mail’, ‘memberof’, ‘sAMAccountName’]
)

if not record:
return {‘status’: False, ‘response’: f”User {user_principal_name} not found in AD”}

user_dn = record[2][0][‘dn’]

# Modify the accountExpires attribute
result = conn.modify(user_dn, {‘accountExpires’: [(MODIFY_REPLACE, [filetime])]})

if result:
return {‘status’: True, ‘response’: f”Account expiry for {user_principal_name} updated successfully.”}
else:
return {‘status’: False, ‘response’: f”Failed to update account expiry: {conn.result}”}

except Exception as e:
return {‘status’: False, ‘response’: str(e)}

def disable_ad_account(user_principal_name):
try:
search_filter = f”(&(sAMAccountName={user_principal_name})(objectclass=person))”
record = conn.search(
‘DC=ratnakarbank,DC=in’,
search_filter,
attributes=[‘cn’, ‘sn’, ‘name’, ’employeeID’, ‘mail’, ‘memberof’, ‘sAMAccountName’]
)
if not record:
return {‘status’: False, ‘response’: f”User {user_principal_name} not found in AD”}

user_dn = record[2][0][‘dn’]
result = conn.modify(user_dn, {‘userAccountControl’: [(MODIFY_REPLACE, [514])]})

if result:
return {‘status’: True, ‘response’: f”User {user_principal_name} disabled successfully”}
else:
return {‘status’: False, ‘response’: f”Failed to disable user {user_principal_name}: {conn.result}”}
except Exception as e:
return {‘status’: False, ‘response’: str(e)}

def enable_ad_account(user_principal_name):
try:
search_filter = f”(&(sAMAccountName={user_principal_name})(objectclass=person))”
record = conn.search(
‘DC=ratnakarbank,DC=in’,
search_filter,
attributes=[‘cn’, ‘sn’, ‘name’, ’employeeID’, ‘mail’, ‘memberof’, ‘sAMAccountName’]
)
if not record:
return {‘status’: False, ‘response’: f”User {user_principal_name} not found in AD”}

user_dn = record[2][0][‘dn’]
result = conn.modify(user_dn, {‘userAccountControl’: [(MODIFY_REPLACE, [512])]})

if result:
return {‘status’: True, ‘response’: f”User {user_principal_name} enabled successfully”}
else:
return {‘status’: False, ‘response’: f”Failed to enable user {user_principal_name}: {conn.result}”}
except Exception as e:
return {‘status’: False, ‘response’: str(e)}

def connect_db():
#cx_Oracle.init_oracle_client(lib_dir=r”/Users/a27536/Downloads/instantclient_23_3″)
conn = cx_Oracle.connect(user=settings.DB_USER, password=settings.DB_PWD, dsn=settings.IMANAGE_DB_DSN, encoding=”UTF-8″)
return conn

try:
query = f”””select im_app_name, im_module_name, im_request_id, im_request_type, B_G_ID, B_G_ID_NAME, DEPARTMENT, CONTACT_NO, OWNER_OF_GID_BID, TYPE_OF_GID_BID, TEMPORARY_OR_PERMANENT, IF_TEMP_DATE_OF_ID_DELETED, HOSTNAME, RETRY_COUNTER, REQUEST_FOR_ID, TYPE_OF_REQUEST, NEW_OWNER_ID, ACCOUNT_END_DATE, B_G_EMAIL_ID from gid_bid where (bot_status is null or bot_status=’Failed’) and retry_counter < 3 “””

im_conn = connect_db()
im_cursor = im_conn.cursor()
im_cursor.execute(query)
result = im_cursor.fetchall()
conn = ldap_connection()

for i in result :
im_app_name = i[0]
IM_MODULE_NAME = i[1]
im_request_id = i[2]
im_request_type = i[3]
B_G_ID = i[4]
B_G_ID_NAME = i[5]
DEPARTMENT = i[6]
CONTACT_NO = i[7]
OWNER_OF_GID_BID = i[8]
TYPE_OF_GID_BID = i[9]
TEMPORARY_OR_PERMANENT = i[10]
IF_TEMP_DATE_OF_ID_DELETED = i[11]
HOSTNAME = i[12]
RETRY_COUNTER = i[13]
REQUEST_FOR_ID = i[14]
TYPE_OF_REQUEST = i[15]
NEW_OWNER_ID = i[16]
ACCOUNT_END_DATE = i[17]
B_G_EMAIL_ID = i[18]
status = ”
ID_TYPE = ”
if im_app_name == ‘Generic Id’:
ID_TYPE=’GENERIC’
elif im_app_name == ‘Bot Id’:
ID_TYPE=’BOT’
elif im_app_name == ‘GIFT Id’:
ID_TYPE=’GIFT’

to_recipients =[]
ccRecipients = []
req_obj = IM_REQUEST.objects.filter(im_request_id=im_request_id).first()
request_for_id = req_obj.request_for_id
requester_id = req_obj.requester_id
business_justification = req_obj.business_justification
requester_email1 = ”
local_part = OWNER_OF_GID_BID.split(‘@’, 1)[0].strip()
emp_obj3 = (EmployeeDetail.objects.filter(employee_email__istartswith=f”{local_part}@”).first())
requesterfor_name = emp_obj3.employee_name
requesterfor_email11 = emp_obj3.employee_email
requester_department = emp_obj3.department
manager_name_for_dn = emp_obj3.manager_name
if requesterfor_email11:
requesterfor_email1 = requesterfor_email11.strip()
to_recipients.append(OWNER_OF_GID_BID)

if ID_TYPE:
if im_request_type == ‘Creation’:
status = generic_bot_gift_id_creation(B_G_ID_NAME ,DEPARTMENT , CONTACT_NO , OWNER_OF_GID_BID , TYPE_OF_GID_BID , TEMPORARY_OR_PERMANENT , IF_TEMP_DATE_OF_ID_DELETED , HOSTNAME, ID_TYPE, IM_MODULE_NAME, REQUEST_FOR_ID,manager_name_for_dn)
password = status[‘password’]
new_password = base64.b64encode(password.encode()).decode() or ”
B_G_ID =”
if status[‘status’] == True :
B_G_ID = status[‘response’]
bot_status= ‘Success’
bot_errormsg=’User created and configured successfully’
bot_errortype=”
else :
bot_status= ‘Failed’
bot_errormsg = status[‘response’]
bot_errortype=”
print(‘bot_errormsg–>’,bot_errormsg, B_G_ID, new_password, bot_status)
query=f”””UPDATE gid_bid SET updated_by=’CRON’, bot_status='{bot_status}’, bot_errormsg='{bot_errormsg}’, bot_errortype ='{bot_errortype}’, B_G_ID='{B_G_ID}’, USER_PASSWORD = ‘{new_password}’ where im_request_id='{im_request_id}'”””

print(‘query..’, query)
try:
im_cursor.execute(query)
except Exception as e:
update_log(req=”, res=e, err_type=’T’, error_code=”,error_desc=f'{e} – Error for Query = {query}’, error_source=’GID_BID_QUERY_ALERT’,res_time=datetime.now(), tp_req=”, service_name=”, file_name=”,cron_stime=”, sh_name=’GID_BID_BACKEND.sh’)
im_conn.commit()

if bot_status == ‘Success’:
subject = f'{IM_MODULE_NAME} Created Successfully! – [Request Id: {im_request_id}]’
if “Generic Id With Email” in IM_MODULE_NAME:
context={
‘username’:requesterfor_name,
‘im_request_id’:im_request_id,
‘generic_id’:B_G_ID,
‘generic_name’:B_G_ID_NAME,
‘generic_email’:B_G_EMAIL_ID,
‘generic_pass’:password,
}
email_body = render_to_string(‘/app/RAM/MASTER/templates/email_templates/generic_id_creation_with_email.html’, context)

if “Generic Id Without Email” in IM_MODULE_NAME:
context={
‘username’:requesterfor_name,
‘im_request_id’:im_request_id,
‘generic_id’:B_G_ID,
‘generic_name’:B_G_ID_NAME,
‘generic_pass’:password,
}
email_body = render_to_string(‘/app/RAM/MASTER/templates/email_templates/generic_id_creation_without_email.html’, context)

if “Bot Id” in IM_MODULE_NAME:
# print(“bot Suc”)
context={
‘username’:requesterfor_name,
‘im_request_id’:im_request_id,
‘bot_id’:B_G_ID,
‘bot_name’:B_G_ID_NAME,
‘bot_email’:B_G_EMAIL_ID,
‘bot_pass’:password,
}
email_body = render_to_string(‘/app/RAM/MASTER/templates/email_templates/bot_id_creation_email.html’, context)

if “GIFT Id” in IM_MODULE_NAME:
context={
‘username’:requesterfor_name,
‘im_request_id’:im_request_id,
‘gift_userid’:B_G_ID,
‘gift_username’:B_G_ID_NAME,
‘gift_pass’:password,
}
email_body = render_to_string(‘/app/RAM/MASTER/templates/email_templates/gift_id_creation_email.html’, context)

response = gapi.main(email_body, subject, to_recipients,ccRecipients)

# Handle response
if ‘Error sending email’ in response:
req = f”{im_request_id} – Mail not sent for {IM_MODULE_NAME} “
update_log(req=req, res=str(response), err_type=’T’, error_code=str(response),error_desc=’Failed to send email to manager/requester’, error_source=’GRAPH_API’,res_time=generate_time_stamp(), tp_req=generate_time_stamp(),service_name=’SEND_EMAIL_TO_GID_BID_CRON’, file_name=’gid_bid_backend.py’,cron_stime=generate_time_stamp(), sh_name=’SEND_EMAIL_TO_GID_BID_CRON’)

if ‘Email sent successfully!’ in response:
req = f”{im_request_id} – Mail sent for {IM_MODULE_NAME}”
update_log(req=req, res=str(response), res_time=generate_time_stamp(),tp_req=generate_time_stamp(), service_name=’SEND_EMAIL_TO_GID_BID_CRON’,file_name=’gid_bid_backend.py’, cron_stime=generate_time_stamp(),sh_name=’SEND_EMAIL_TO_GID_BID_CRON’)

update_is_active(request_for_id,’GID_BID’,1)

if im_request_type == ‘Modification’:
subject = ”
if TYPE_OF_REQUEST == “Password Reset”:
reset_status = reset_ad_password(B_G_ID)
password = reset_status[‘password’]
new_password = base64.b64encode(password.encode()).decode() or ”
bot_status = ‘Success’ if reset_status[‘status’] else ‘Failed’
bot_errormsg = f”{reset_status[‘status’]} – {reset_status[‘response’]}”
# print(bot_errormsg)
bot_errortype = ”
query = f”””UPDATE gid_bid SET updated_by=’CRON’, bot_status='{bot_status}’, bot_errormsg='{bot_errormsg}’, bot_errortype='{bot_errortype}’, USER_PASSWORD = ‘{new_password}’ WHERE im_request_id='{im_request_id}'”””
try:
im_cursor.execute(query)
except Exception as e:
update_log(req=”, res=e, err_type=’T’, error_code=”,error_desc=f'{e} – Error for Query = {query}’, error_source=’GID_BID_QUERY_ALERT’,res_time=datetime.now(), tp_req=”, service_name=”, file_name=”,cron_stime=”, sh_name=’GID_BID_BACKEND.sh’)
im_conn.commit()
#email sent logic written below.
if bot_status == ‘Success’:
subject = f'[iManage: Password Reset of ID Completed â€“ {im_request_id}]’
context={
‘username’:requesterfor_name,
‘im_request_id’:im_request_id,
‘type_of_req’:TYPE_OF_REQUEST,
‘required_value’:password,
‘gift_userid’:B_G_ID,
}
else:
context = ‘No’
elif TYPE_OF_REQUEST == “Change Ownership”:
emp_obj = EmployeeDetail.objects.filter(emp_id__iexact = NEW_OWNER_ID).first()
employee_name = emp_obj.employee_name
new_owner_dn=fetch_user_dn(NEW_OWNER_ID)
if new_owner_dn == False:
error_message = f’Manager DN is null for {NEW_OWNER_ID} ID creation’
res_datetime = datetime.now()
update_log(req=”, res=error_message, err_type=’T’, error_code=”,error_desc=str(error_message), error_source=f'{ID_TYPE} ID CREATION’,res_time=res_datetime, tp_req=”, service_name=”,file_name=”, cron_stime=”, sh_name=’AD_ID_CREATION.sh’)
bot_status = ‘Failed’
bot_errormsg = f”{error_message}”
bot_errortype = ”
query = f”””UPDATE gid_bid SET updated_by=’CRON’, bot_status='{bot_status}’, bot_errormsg='{bot_errormsg}’, bot_errortype='{bot_errortype}’ WHERE im_request_id='{im_request_id}'”””
try:
im_cursor.execute(query)
except Exception as e:
update_log(req=”, res=e, err_type=’T’, error_code=”,error_desc=f'{e} – Error for Query = {query}’, error_source=’GID_BID_QUERY_ALERT’,res_time=datetime.now(), tp_req=”, service_name=”, file_name=”,cron_stime=”, sh_name=’GID_BID_BACKEND.sh’)
im_conn.commit()
context = ‘No’
else:
changeOwner_status = change_ad_owner(B_G_ID,new_owner_dn,employee_name)
bot_status = ‘Success’ if changeOwner_status[‘status’] else ‘Failed’
bot_errormsg = f”{changeOwner_status[‘status’]} – {changeOwner_status[‘response’]}”
bot_errortype = ”
query = f”””UPDATE gid_bid SET updated_by=’CRON’, bot_status='{bot_status}’, bot_errormsg='{bot_errormsg}’, bot_errortype='{bot_errortype}’ WHERE im_request_id='{im_request_id}'”””
try:
im_cursor.execute(query)
except Exception as e:
update_log(req=”, res=e, err_type=’T’, error_code=”,error_desc=f'{e} – Error for Query = {query}’, error_source=’GID_BID_QUERY_ALERT’,res_time=datetime.now(), tp_req=”, service_name=”, file_name=”,cron_stime=”, sh_name=’GID_BID_BACKEND.sh’)
im_conn.commit()
#email sent logic written below.
if bot_status == ‘Success’:
subject = f'[iManage: Change of Owner Completed â€“ {im_request_id}]’
required_value = f'{employee_name} ({NEW_OWNER_ID})’
context={
‘username’:requesterfor_name,
‘im_request_id’:im_request_id,
‘type_of_req’:TYPE_OF_REQUEST,
‘required_value’:required_value,
‘gift_userid’:B_G_ID,
}
else:
context = ‘No’
elif TYPE_OF_REQUEST == “Id Extension Validity”:
contact_status = update_ad_account_expiry(B_G_ID, ACCOUNT_END_DATE)
bot_status = ‘Success’ if contact_status[‘status’] else ‘Failed’
bot_errormsg = f”{contact_status[‘status’]} – {contact_status[‘response’]}”
# print(bot_errormsg)
bot_errortype = ”
query = f”””UPDATE gid_bid SET updated_by=’CRON’, bot_status='{bot_status}’, bot_errormsg='{bot_errormsg}’, bot_errortype='{bot_errortype}’ WHERE im_request_id='{im_request_id}'”””
try:
im_cursor.execute(query)
except Exception as e:
update_log(req=”, res=e, err_type=’T’, error_code=”,error_desc=f'{e} – Error for Query = {query}’, error_source=’GID_BID_QUERY_ALERT’,res_time=datetime.now(), tp_req=”, service_name=”, file_name=”,cron_stime=”, sh_name=’GID_BID_BACKEND.sh’)
im_conn.commit()
#email sent logic written below.
if bot_status == ‘Success’:
subject = f'[iManage: ID Extension Validity Completed â€“ {im_request_id}]’
context={
‘username’:requesterfor_name,
‘im_request_id’:im_request_id,
‘type_of_req’:TYPE_OF_REQUEST,
‘required_value’:ACCOUNT_END_DATE,
‘gift_userid’:B_G_ID,
}
else:
context = ‘No’
elif TYPE_OF_REQUEST == “Mobile Number Change”:
contact_status = update_ad_mobile_number(B_G_ID, CONTACT_NO)
bot_status = ‘Success’ if contact_status[‘status’] else ‘Failed’
bot_errormsg = f”{contact_status[‘status’]} – {contact_status[‘response’]}”
# print(bot_errormsg)
bot_errortype = ”
query = f”””UPDATE gid_bid SET updated_by=’CRON’, bot_status='{bot_status}’, bot_errormsg='{bot_errormsg}’, bot_errortype='{bot_errortype}’ WHERE im_request_id='{im_request_id}'”””
try:
im_cursor.execute(query)
except Exception as e:
update_log(req=”, res=e, err_type=’T’, error_code=”,error_desc=f'{e} – Error for Query = {query}’, error_source=’GID_BID_QUERY_ALERT’,res_time=datetime.now(), tp_req=”, service_name=”, file_name=”,cron_stime=”, sh_name=’GID_BID_BACKEND.sh’)
im_conn.commit()
#email sent logic written below.
if bot_status == ‘Success’:
subject = f'[iManage: Mobile Number Change Completed â€“ {im_request_id}]’
context={
‘username’:requesterfor_name,
‘im_request_id’:im_request_id,
‘type_of_req’:TYPE_OF_REQUEST,
‘required_value’:CONTACT_NO,
‘gift_userid’:B_G_ID,
}
else:
context = ‘No’
elif TYPE_OF_REQUEST in [“Interactive to Non-Interactive”,”Non-Interactive to Interactive”]:
gBid_status = update_id_types(B_G_ID,HOSTNAME)
bot_status = ‘Success’ if gBid_status[‘status’] else ‘Failed’
bot_errormsg = f”{gBid_status[‘status’]} – {gBid_status[‘response’]}”
bot_errortype = ”
query = f”””UPDATE gid_bid SET updated_by=’CRON’, bot_status='{bot_status}’, bot_errormsg='{bot_errormsg}’, bot_errortype='{bot_errortype}’ WHERE im_request_id='{im_request_id}'”””
try:
im_cursor.execute(query)
except Exception as e:
update_log(req=”, res=e, err_type=’T’, error_code=”,error_desc=f'{e} – Error for Query = {query}’, error_source=’GID_BID_QUERY_ALERT’,res_time=datetime.now(), tp_req=”, service_name=”, file_name=”,cron_stime=”, sh_name=’GID_BID_BACKEND.sh’)
im_conn.commit()
#email sent logic written below.
if bot_status == ‘Success’:
subject = f'[iManage: Interactive ID to Non-Interactive ID Completed â€“ {im_request_id}]’ if TYPE_OF_REQUEST == “Interactive to Non-Interactive” else f'[iManage: Non-Interactive ID to Interactive ID Completed â€“ {im_request_id}]’
required_value = ‘Non-Interactive’ if TYPE_OF_REQUEST == “Interactive to Non-Interactive” else ‘Interactive’
context={
‘username’:requesterfor_name,
‘im_request_id’:im_request_id,
‘type_of_req’:TYPE_OF_REQUEST,
‘required_value’:required_value,
‘gift_userid’:B_G_ID,
}
else:
context = ‘No’

if ‘success’ in bot_status.lower():
if context != ‘No’:
email_body = render_to_string(‘/app/RAM/MASTER/templates/email_templates/gid_bid_modification_resolved_notification.html’, context)
response = gapi.main(email_body, subject, to_recipients,ccRecipients)
# Handle response
if ‘Error sending email’ in response:
req = f”{im_request_id} – Mail not sent for {IM_MODULE_NAME} “
update_log(req=req, res=str(response), err_type=’T’, error_code=str(response),error_desc=’Failed to send email to manager/requester’, error_source=’GRAPH_API’,res_time=generate_time_stamp(), tp_req=generate_time_stamp(),service_name=’SEND_EMAIL_TO_GID_BID_CRON’, file_name=’gid_bid_backend.py’,cron_stime=generate_time_stamp(), sh_name=’SEND_EMAIL_TO_GID_BID_CRON’)

if ‘Email sent successfully!’ in response:
req = f”{im_request_id} – Mail sent for {IM_MODULE_NAME} “
update_log(req=req, res=str(response), res_time=generate_time_stamp(),tp_req=generate_time_stamp(), service_name=’SEND_EMAIL_TO_GID_BID_CRON’,file_name=’gid_bid_backend.py’, cron_stime=generate_time_stamp(),sh_name=’SEND_EMAIL_TO_GID_BID_CRON’)

update_is_active(request_for_id,’GID_BID’,1)

if im_request_type == ‘Deletion / Disable’:
disable_status = disable_ad_account(B_G_ID)
bot_status = ‘Success’ if disable_status[‘status’] else ‘Failed’
bot_errormsg = f”{disable_status[‘status’]} – {disable_status[‘response’]}”
# print(bot_errormsg)
bot_errortype = ”
query = f”””UPDATE gid_bid SET updated_by=’CRON’, bot_status='{bot_status}’, bot_errormsg='{bot_errormsg}’, bot_errortype='{bot_errortype}’ WHERE im_request_id='{im_request_id}'”””
try:
im_cursor.execute(query)
except Exception as e:
update_log(req=”, res=e, err_type=’T’, error_code=”,error_desc=f'{e} – Error for Query = {query}’, error_source=’GID_BID_QUERY_ALERT’,res_time=datetime.now(), tp_req=”, service_name=”, file_name=”,cron_stime=”, sh_name=’GID_BID_BACKEND.sh’)
im_conn.commit()
if bot_status == ‘Success’:
subject = f'{IM_MODULE_NAME} Disabled Successfully! – [Request Id: {im_request_id}]’
context={
‘username’:requesterfor_name,
‘im_request_id’:im_request_id,
‘gift_bot_gen_id’:im_app_name,
‘gid_bid_id’:B_G_ID,
‘gid_bid_name’:B_G_ID_NAME,
}
email_body = render_to_string(‘/app/RAM/MASTER/templates/email_templates/gid_bid_disableId_email.html’, context)

response = gapi.main(email_body, subject, to_recipients,ccRecipients)
# Handle response
if ‘Error sending email’ in response:
req = f”{im_request_id} – Mail not sent for {IM_MODULE_NAME} “
update_log(req=req, res=str(response), err_type=’T’, error_code=str(response),error_desc=’Failed to send email to manager/requester’, error_source=’GRAPH_API’,res_time=generate_time_stamp(), tp_req=generate_time_stamp(),service_name=’SEND_EMAIL_TO_GID_BID_CRON’, file_name=’gid_bid_backend.py’,cron_stime=generate_time_stamp(), sh_name=’SEND_EMAIL_TO_GID_BID_CRON’)

update_is_active(request_for_id,’GID_BID’,0)

if im_request_type == ‘Enable’:
enable_status = enable_ad_account(B_G_ID)
bot_status = ‘Success’ if enable_status[‘status’] else ‘Failed’
bot_errormsg = f”{enable_status[‘status’]} – {enable_status[‘response’]}”
# print(bot_errormsg)
bot_errortype = ”
query = f”””UPDATE gid_bid SET updated_by=’CRON’, bot_status='{bot_status}’, bot_errormsg='{bot_errormsg}’, bot_errortype='{bot_errortype}’ WHERE im_request_id='{im_request_id}'”””
try:
im_cursor.execute(query)
except Exception as e:
update_log(req=”, res=e, err_type=’T’, error_code=”,error_desc=f'{e} – Error for Query = {query}’, error_source=’GID_BID_QUERY_ALERT’,res_time=datetime.now(), tp_req=”, service_name=”, file_name=”,cron_stime=”, sh_name=’GID_BID_BACKEND.sh’)
im_conn.commit()
if bot_status == ‘Success’:
subject = f'{IM_MODULE_NAME} Enabled Successfully! – [Request Id: {im_request_id}]’
context={
‘username’:requesterfor_name,
‘im_request_id’:im_request_id,
‘gift_bot_gen_id’:im_app_name,
‘gid_bid_id’:B_G_ID,
‘gid_bid_name’:B_G_ID_NAME,
}
email_body = render_to_string(‘/app/RAM/MASTER/templates/email_templates/gid_bid_enableId_email.html’, context)

update_is_active(request_for_id,’GID_BID’,1)

except Exception as e:
# import traceback
# traceback.print_exc()
update_log(req=”, res=e, err_type=’T’, error_code=”,error_desc=str({e}), error_source=’GID_BID_ENTIRE_CODE_ALERT’,res_time=datetime.now(), tp_req=”, service_name=”, file_name=”,cron_stime=”, sh_name=’GID_BID_BACKEND.sh’)
pass