我有一个函数应用程序,它尝试使用 azure.identity python 库中名为 MI_ODBC 的用户分配的托管标识资源和 DefaultAzureCredential 来将
[1234, 'Hello']
插入到名为 test_table_do_not_use
的 SQL 表中,以进行身份验证,即无需凭据即可进行身份验证。之前,我能够按照此microsoft 指南为我们的 Azure SQL 数据库成功执行此操作,但在切换到 Azure SQL 托管实例后,我现在遇到了问题。功能应用程序代码:
import os
import pyodbc, struct
from azure.identity import DefaultAzureCredential
import azure.functions as func
import logging
sql_insert_data = [1234, 'Hello']
def insert_all(data):
sql = "INSERT INTO database.dbo.test_table_do_not_use(column1, column2) VALUES (?, ?)"
cnxn = get_conn()
crsr = cnxn.cursor()
try:
crsr.execute(sql, data)
except Exception as e:
crsr.rollback()
print(e)
print('Transaction rollback')
else:
cnxn.commit()
crsr.close()
cnxn.close()
# Managed Identity Auth
def get_conn():
credential = DefaultAzureCredential(exclude_interactive_browser_credential=False)
token_bytes = credential.get_token("https://database.windows.net/.default").token.encode("UTF-16-LE")
token_struct = struct.pack(f'<I{len(token_bytes)}s', len(token_bytes), token_bytes)
SQL_COPT_SS_ACCESS_TOKEN = 1256 # This connection option is defined by microsoft in msodbcsql.h
connection_string = os.environ["SQLAZURECONNSTR_connectionstringname"]
conn = pyodbc.connect(connection_string, attrs_before={SQL_COPT_SS_ACCESS_TOKEN: token_struct})
return conn
app = func.FunctionApp()
@app.schedule(schedule="0 */1 * * * *", arg_name="myTimer", run_on_startup=True,
use_monitor=False)
def timer_trigger(myTimer: func.TimerRequest) -> None:
if myTimer.past_due:
logging.info('The timer is past due!')
logging.info('Python timer trigger function executed.')
return insert_all(data=sql_insert_data)
这是我一直用于用户分配的托管标识/DefaultAzureCredential 身份验证的连接字符串:
Driver={ODBC Driver 18 for SQL Server};Server=tcp:<database-server-name>.database.windows.net,1433;Database=<database-name>;Encrypt=yes;TrustServerCertificate=no;Connection Timeout=30
但是,我不断收到此错误:
Result: Failure Exception: InterfaceError: ('28000', "[28000] [Microsoft][ODBC Driver 18 for SQL Server][SQL Server]Login failed for user '<token-identified principal>'. (18456) (SQLDriverConnect)") Stack: File "/azure-functions-host/workers/python/3.11/LINUX/X64/azure_functions_worker/dispatcher.py", line 505, in _handle__invocation_request call_result = await self._loop.run_in_executor( ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/usr/local/lib/python3.11/concurrent/futures/thread.py", line 58, in run result = self.fn(*self.args, **self.kwargs) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/azure-functions-host/workers/python/3.11/LINUX/X64/azure_functions_worker/dispatcher.py", line 778, in _run_sync_func return ExtensionManager.get_sync_invocation_wrapper(context, ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/azure-functions-host/workers/python/3.11/LINUX/X64/azure_functions_worker/extension.py", line 215, in _raw_invocation_wrapper result = function(**args) ^^^^^^^^^^^^^^^^ File "/home/site/wwwroot/function_app.py", line 73, in timer_trigger return insert_all(data=sql_insert_data) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/home/site/wwwroot/function_app.py", line 30, in insert_all cnxn = get_conn() ^^^^^^^^^^ File "/home/site/wwwroot/function_app.py", line 62, in get_conn conn = pyodbc.connect(connection_string, attrs_before={SQL_COPT_SS_ACCESS_TOKEN: token_struct})
如果我使用 sql 身份验证以及存储在函数应用程序的配置 -> 连接字符串中的凭据,然后使用以下代码调用环境变量,则可以使函数应用程序正常工作:
def get_conn():
connection_string = os.environ["SQLAZURECONNSTR_sqlauthstringname"]
conn = pyodbc.connect(connection_string)
return conn
我将用户分配的托管标识添加到函数应用的“身份”部分和托管实例中,我还使用以下 SQL 代码将托管标识添加为 Azure SQL MI 数据库中的用户:
CREATE USER [user-assigned-identity-name] FROM EXTERNAL PROVIDER;
ALTER ROLE db_datareader ADD MEMBER [user-assigned-identity-name];
ALTER ROLE db_datawriter ADD MEMBER [user-assigned-identity-name];
ALTER ROLE db_ddladmin ADD MEMBER [user-assigned-identity-name];
用户分配的托管标识具有托管实例所在资源组的角色分配 SQL 托管实例贡献者。
最后,我还在托管实例的网络安全组资源的入站安全规则中添加了“AzureCloud”(优先级设置为400),以将所有函数应用列入白名单。
如果有人对为什么我不断收到用户分配的托管标识的身份验证错误有任何想法 - 就像我在为 sql 托管实例设置托管标识时没有考虑到的一些考虑因素一样,我的连接字符串存在问题等等
答案是将用户添加到我正在连接的特定数据库中。我在连接到数据库 1 时使用了更改角色代码,而我应该连接到数据库 2,或者我应该在更改角色代码之前编写 USE 数据库 2。将来我将使用来自 mssql Tips
的此代码的修改版本Use master
GO
DECLARE @dbname VARCHAR(50)
DECLARE @statement NVARCHAR(max)
DECLARE db_cursor CURSOR
LOCAL FAST_FORWARD
FOR
SELECT name
FROM MASTER.dbo.sysdatabases
WHERE name NOT IN ('master','model','msdb','tempdb','distribution')
OPEN db_cursor
FETCH NEXT FROM db_cursor INTO @dbname
WHILE @@FETCH_STATUS = 0
BEGIN
SELECT @statement = 'use '+@dbname +';'+ 'CREATE USER [TipsDemoUser]
FOR LOGIN [TipsDemoUser]; EXEC sp_addrolemember N''db_datareader'',
[TipsDemoUser];EXEC sp_addrolemember N''db_datawriter'', [TipsDemoUser]'
exec sp_executesql @statement
FETCH NEXT FROM db_cursor INTO @dbname
END
CLOSE db_cursor
DEALLOCATE db_cursor
我说修改是因为 sp_addrolemember 存储过程已被弃用,正如 microsoft 在这里指出的那样,“此功能 [sp_addrolemember] 将在 SQL Server 的未来版本中删除。避免在新的开发工作中使用此功能,并计划修改应用程序当前正在使用此功能。请改用 ALTER ROLE。”
显然有一种方法可以添加服务器级别的角色分配,以便所有数据库都继承角色和权限。如果可以的话,那会比上面MSSQL技巧中的解决方案好得多。