I would like to access a SQL server from windows 10 using python and pyodbc, not with the account i currently am logged in windows but with a different windows account
runan different user cannot work because the user that have access to the DB has not access to python directory nad i can include him in the users of my PC
Is it possible?
So, I ended up solving this not using pyodbc but using pymssql instead. I'm not sure if that help OP though. Just figured I'd share.
import dotenv, os, pymssql, win32security, win32con
from modules.utility import write_to_json
dotenv.load_dotenv()
def impersonate_user(username, password, domain):
"""Impersonate the security context of another user."""
handler = win32security.LogonUser(
username, domain, password,
win32con.LOGON32_LOGON_INTERACTIVE,
win32con.LOGON32_PROVIDER_DEFAULT)
win32security.ImpersonateLoggedOnUser(handler)
write_to_json(get_records_from_database(), 'JSON_FILE_NAME')
handler.Close()
def terminate_impersonation():
"""Terminate the impersonation of another user."""
win32security.RevertToSelf()
def connect_pymssql():
con = pymssql.connect(
server=os.getenv('DB_SERVER'),
database=os.getenv('DB_NAME'),
port=os.getenv('DB_PORT')
)
cur = con.cursor()
return con, cur
def get_records_from_database():
con, cur = connect_pymssql()
cur.execute(f"select * from [TABLE];")
result = cur.fetchall()
con.close()
return [list(x) for x in result] if result else None
Actually the way proposed by Tim Griffith on 1st April 2022 works pretty well also with pyodbc (so it's suitable as solution of the topic). You just should consider the error codes and results returned by Windows API before further processing to avoid error messages like one named in comment to Tim's answer. I've tested it on Windows 10, here is the complete code:
import win32api
import win32security
import win32con
import pyodbc
# tested with python 3.8.7 and pyodbc 4.0.30
# Hint: You should resolve all "TODO" in the script first.
# TODO: set proper values of global variables below:
global_domain = 'TODO_DOMAIN'
global_user = 'TODO_USER'
global_pass = 'TODO_PASS'
global_ODBC_driver = '{SQL Server}' # TODO: the driver name might differ on your side (check ODBC configuration and correct the name here if needed)
global_ODBC_server = 'TODO_SERVER'
global_ODBC_database = 'TODO_DATABASE'
global_ODBC_schema = 'TODO_SCHEMA'
global_ODBC_table = 'TODO_TABLE'
global_ODBC_column_1 = 'TODO_COLUMN_1'
global_ODBC_column_2 = 'TODO_COLUMN_2'
global_ODBC_criteria = ' WHERE TODO_COLUMN_NAME = \'TODO_VALUE\''
# Converts given "usual" string in_str to a raw-string
# Note: raw-string is a string which was defined with prefix r or R (i.e. r'te\st') and handles escape characters "as is" (i.e. not trying to "process" them in between)
def strToRawStr(in_str):
return str(in_str).encode('unicode_escape').decode()
# "impersonate" authentification data and security context of user defined by domain, user and pwd
# returns handler of User
def impersonate_user(domain, user, pwd):
if None == user or None == domain or None == pwd:
print('impersonate_user(): Error: some of arguments is empty!')
return (False, None)
domain = str(domain)
user = str(user)
pwd = str(pwd)
result = None
print('impersonate_user(): Starting impersonation of user "' + user + '"')
handler = None
try:
handler = win32security.LogonUser(user, domain, pwd, win32con.LOGON32_LOGON_INTERACTIVE, win32con.LOGON32_PROVIDER_DEFAULT)
except Exception as e: # in case the domain/user/password combination is not correct - an exception will happen
print('impersonate_user(): Exception: ' + str(e))
if None != handler and 0 != handler:
result = win32security.ImpersonateLoggedOnUser(handler)
if None == result or 0 == result:
print('impersonate_user(): Last error code:' + str(win32api.GetLastError()))
# Note: you might need to add proper processing of result and error code here (i.e. in case of error -
# instead of continue processing - call handler.Close() and return handler = None)
print('impersonate_user(): result=' + str(result) + ', handler=' + str(handler))
return (result, handler)
# closes impersonation of user with given handler and resets "local" windows security environment to "own" user
def close_impersonation(handler):
print('close_impersonation(): Closing impersonation of different user')
if None != handler:
handler.Close()
win32security.RevertToSelf()
print('close_impersonation(): Done.')
# check the DB connection (just connect, get limited amount of rows, print how many rows fetched and close the connection).
def check_DB_connection():
global global_ODBC_driver, global_ODBC_server, global_ODBC_database, global_ODBC_schema
global global_ODBC_table, global_ODBC_column_1, global_ODBC_column_2, global_ODBC_criteria
conn_str = (
r'DRIVER=' + strToRawStr(global_ODBC_driver) + r';'
r'SERVER=' + strToRawStr(global_ODBC_server) + r';'
r'DATABASE=' + strToRawStr(global_ODBC_database) + r';'
r'Integrated_Security=SSPI;' # Enabling the SSPI security mechanism
r'Trusted_Connection=yes;' # I think it should be yes ;-)
)
# connect to DB and create pyodbc "cursor" (i.e. session)
print('Connecting to DB...')
conn = pyodbc.connect(conn_str) # connection
print('Creating session to DB...')
cursor = conn.cursor() # session (i.e. cursor)
print('Session created, executing request(s)...')
data_source = global_ODBC_database + '.' + global_ODBC_schema + '.' + global_ODBC_table
cursor.execute('SELECT "' + global_ODBC_column_1 + '", "' + global_ODBC_column_2 + '" ' \
+ 'FROM ' + data_source + global_ODBC_criteria)
print('Fetching up to 1000 results ...')
rows = cursor.fetchmany(1000) # get up to 1000 results at once
print('\nFetched (' + str(len(rows)) + ' rows).')
# Note: if count is bigger than 0 - you can enumerate here rows in usual way (see pyodbc examples)
conn.close()
print('Connection to DB closed.')
###################
## main function ##
###################
def main():
# TODO: set the global variables above first and then remove next 2 lines:
print('First set at top of the script the proper global variables values, then try!')
exit(1)
print('Started.')
global global_domain, global_user, global_pass
(r, h) = impersonate_user(global_domain, global_user, global_pass)
if None != h:
check_DB_connection()
else:
print('Error: Can\'t impersonate given user, exiting without DB check.')
close_impersonation(h)
print('Finished.')
# script entry-point for main-function:
if __name__ == '__main__':
main()
