kale1314 / lamda Goto Github PK

from lamda.client import *

d = Device("192.168.0.2")
# 如果在服务端启用了 certificate 请这样连接
d = Device("192.168.0.2", certificate="/path/to/lamda.pem")

profile = GproxyProfile()
profile.type = GproxyType.HTTP_CONNECT

# 此选项请根据实际情况选择你是否需要
profile.drop_udp = True
profile.host = "123.123.123.123"
profile.port = 8080

d.start_gproxy(profile)

profile = GproxyProfile()
# socks5 代理则为 GproxyType.SOCKS5
profile.type = GproxyType.HTTP_CONNECT
# 如果你需要重定向 DNS 查询到 114.114.114.114
# 注意此 DNS 是系统全局的，系统发出的所有DNS将会被转发
# 如果是与OpenVPN共存的情况，不要设置为OpenVPN的内网DNS服务器，否则可能会导致彻底断网
# 去掉 nameserver 配置行将使用系统默认 DNS
#
# 为什么有此选项：你可以修改一些应用的 dns 域名解析
profile.nameserver = "114.114.114.114"
profile.host = "代理服务器地址"
profile.port = 代理服务器端口

# 如果这个代理服务器需要登录信息（注意：如果没有，请置为 None 或者删除这两行）
profile.password = "代理服务器登录密码"
profile.login = "代理服务器登录用户"

# socks5 模式支持 udp 代理，但是 http 代理并不支持
# 因为 udp 多数情况下并不会被代理，所以禁用 udp 流量是一个不错的选择
# 当 drop_udp 为 True 时，应用/系统的 UDP 流量将会被屏蔽，默认为 False
profile.drop_udp = False

# 本地流量是否需要*不经过*代理，如果为 True，本地流量
# 如 192.168.x.x 10.x.x.x 等路由器内网网段的流量将不会经过代理，默认为 False
# 注意：如果开启了 udp_proxy，此选项对于 UDP 流量无效
profile.bypass_local_subnet = True

# 是否需要代理 udp 流量
# 注意，http 代理服务不支持代理 udp 协议，开启此选项必须使用 socks5 作为代理服务器
# (GproxyType.SOCKS5)，且 socks5 代理服务器必须配置开启 udp 代理模式，
# 需要稍加复杂的服务器配置，为了避坑，请参照 tools/socks5 里的介绍进行安装配置
# 当你使用 http 代理或者 drop_udp 选项为 True，此设置将会被忽略
profile.udp_proxy = False

# 如果需要仅对特定应用使用代理（例如安卓浏览器，如果是全局则删除这两行）
app = d.application("com.android.browser")
profile.application.set(app)

#     注意事项以及提示：
# 设置代理后，正在运行的应用是不会立即使用设置的代理的
# 因为这些应用在设置代理之前就已经完成了 tcp 连接的建立
# 所以，需要你手动关闭应用并启动，应用才会通过代理建立连接
# 也就是说，如果你是做中间人流量分析，那设置代理后
# 你需要关闭应用再重新打开才会看到应用的请求
#
# 注：本机的 DNS 流量始终不会经过代理

# 启动代理
d.start_gproxy(profile)
# 关闭代理
d.stop_gproxy()

import os

# 拼接 mitmproxy-ca-cert.pem 文件的路径
HOME = os.path.expanduser("~")
cert_path = os.path.join(HOME, ".mitmproxy", "mitmproxy-ca-cert.pem")

# 注意这里如果出现 Read-only system partition，如果出现异常，请检查你是否已完成
# 上面的 remount 操作（不论是否报错，请在本文档搜索 remount）
# 如果已完成仍然出现，请检查是否使用的最新版 root 软件 (magisk, supersu)
# 如果仍然没问题，那可能是设备的 system 分区被锁死只读挂载
# 这出现在部分 pixel 设备的系统上，有条件可以尝试降级到安卓7
# 也可手动安装此证书，但这可能会损失一些 lamda 的描述能力
#
# 以 mitmproxy 为例，使用如下代码安装证书
d.install_ca_certificate(cert_path)

# 使用如下代码卸载证书（如不常变化不建议频繁安装卸载）
d.uninstall_ca_certificate(cert_path)
# 此证书安装接口是通用的，你可以用它安装任何应用要求你安装的证书
# 你同样可以用其安装 Fiddler/Charles 要求你安装的证书
# 只需要提供文件路径即可

profile = OpenVPNProfile()

# 是否全局 VPN，为 False 时仅路由服务器端推送的特定网段
profile.all_traffic  = True
# 服务器端开启的连接协议 (或者为 OpenVPNProto.TCP)
profile.proto        = OpenVPNProto.UDP
profile.host         = "OpenVPN 服务器地址"
profile.port         = OpenVPN 服务器端口
# 服务器端通道加密方法
profile.cipher       = OpenVPNCipher.AES_256_GCM

profile.tls_encryption = OpenVPNEncryption.TLS_CRYPT
profile.tls_key_direction = OpenVPNKeyDirection.KEY_DIRECTION_NONE
profile.tls_key = """
-----BEGIN OpenVPN Static key V1-----
tls key / tls auth
-----END OpenVPN Static key V1-----
"""

profile.ca = """
-----BEGIN CERTIFICATE-----
服务端配置的 ca 证书
-----END CERTIFICATE-----
"""

profile.cert = """
-----BEGIN CERTIFICATE-----
客户端证书
-----END CERTIFICATE-----
"""

profile.key = """
-----BEGIN PRIVATE KEY-----
客户端私钥
-----END PRIVATE KEY-----
"""

# 启动 OpenVPN
d.start_openvpn(profile)
# 关闭 OpenVPN
d.stop_openvpn()

# 使用 lamda 时的做法
device = d.frida
device.enumerate_processes()

# 通常做法
manager = frida.get_device_manager()
device = manager.add_remote_device("192.168.0.2:65000")
device.enumerate_processes()

import json
import requests
url = "http://192.168.0.2:65000/fridarpc/myRpcName/getMyString"
# 请求接口
res = requests.post(url, data={"args": json.dumps(["A","B"])})
print (res.status_code, res.json()["result"])

#* 状态码 200 一切正常
#* 状态码 410 需要重新注入脚本或者脚本未注入（目前不支持自动重新注入）
#* 状态码 500 脚本或参数异常
#* 状态码 400 参数错误

from lamda.client import *
# 端口为上面的 rport
d = Device("127.0.0.1", port=12345)
# 浏览器打开 http://127.0.0.1:12345 即可访问远程桌面
# 其余任何接口调用实现均统一，无需做任何改动

# 获取 ro.secure 的值
d.getprop("ro.secure")

# 设置 ro.secure 的值
d.setprop("ro.secure", "0")

settings = d.stub("Settings")

# 示例：获取及修改屏幕亮度
settings.get_system("screen_brightness")
settings.put_system("screen_brightness", "5")

# 示例：关闭开发者选项
settings.get_global("development_settings_enabled")
settings.put_global("development_settings_enabled", "0")

# 示例
settings.get_secure("screensaver_enabled")
settings.put_secure("screensaver_enabled", "0)

status = d.stub("Status")

# 获取设备启动时间
status.get_boot_time()

# 获取设备磁盘使用情况
status.get_disk_usage(mountpoint="/data")

# 获取电池信息
status.get_battery_info()
# 获取CPU使用情况
status.get_cpu_info()
# 获取总体磁盘读写情况
status.get_overall_disk_io_info()
# 获取用户数据磁盘的读写情况 (userdata)
status.get_userdata_disk_io_info()
# 获取总体网络收发情况
status.get_overall_net_io_info()
# 获取 wlan0 接口的网络收发情况
status.get_net_io_info("wlan0")
# 获取内存使用情况
status.get_mem_info()

# 执行前台脚本（执行时间短（0-10秒内）的脚本）
cmd = d.execute_script("whoami")
print (cmd.exitstatus)
print (cmd.stdout)
print (cmd.stderr)

# 执行后台脚本（执行时间长的脚本）
# 对于后台脚本，因考虑可能用户写出死循环脚本无限输出导致内存占满等问题
# 暂时无法获知其执行结果
ID = d.execute_background_script("sleep 100; exit 0;")
# 检查后台脚本是否结束
d.is_background_script_finished(ID)
# 强制结束后台脚本
d.kill_background_script(ID)

debug = d.stub("Debug")

r = debug.set_debuggable()
print (r)

debug = d.stub("Debug")

# 启动 IDA 32 服务端（端口可自定义）
debug.start_ida(port=22032)
# 检查是否已启动
debug.is_ida_running()
# 关闭 IDA 32 服务端
debug.stop_ida()
# 如果调试的是64位程序，将方法名中的 ida 替换为 ida64 即可
# 例如
debug.start_ida64(port=22064)
#
# 如果需要自定义 ida-server 的环境变量例如 IDA_LIBC_PATH (同样适用于 start_ida)
debug.start_ida64(port=22064, IDA_LIBC_PATH="/apex/com.android.runtime/lib64/bionic/libc.so")
# 当你调试的目标程序是32位时使用 start_ida
# 否则使用 start_ida64
# 当你的设备系统为32位平台时，start_ida64 将会无效

# lamda 内置的 adb 服务完全独立于系统本身提供的 adb 服务
# 所以在使用之前需要先手动调用以下接口将你的 adb 公钥安装至设备上
# 否则直接连接将会显示未授权（系统设置开发者模式中授权的秘钥与内置 adb 并不通用）
#
# tools 目录下的 adb_pubkey.py 封装了下面接口的安装过程
# 你可以使用该脚本一键授权，允许本机连接，请查看其 README，以下代码仅做参考说明
#
# 这个秘钥文件位于你电脑上的 ~/.android 或者 C:\\Users\xxxx\.android，文件名为 adbkey.pub
# 如果不存在这个文件但是存在文件 adbkey，请切换到该目录并执行命令
# adb pubkey adbkey >adbkey.pub 来生成 adbkey.pub
#
# 随后使用 python 代码来拼接这个生成的 adbkey.pub 路径
import os
keypath = os.path.join("~", ".android", "adbkey.pub")
abs_keypath = os.path.expanduser(keypath)
print (abs_keypath)
#
# 然后安装这个 adbkey.pub 到 lamda
d.install_adb_pubkey(abs_keypath)
# 这样你就可以连接内置 adb 了
# 通过命令 adb connect 192.168.0.2:65000 连接到设备
# 你完全可以将其理解为 WIFI ADB
#
# 或者如果你需要从 lamda 内置 adb 移除这个公钥
d.uninstall_adb_pubkey(abs_keypath)
#
#
# ** 下面几步都是不必要做的，只是为了介绍存在这些接口 **
#
# 启动设备上的 adb 服务 (默认状态下是随 lamda 启动的，只做演示无需调用)
d.start_android_debug_bridge()
# 停止 adb 服务
d.stop_android_debug_bridge()
# 检查是否启动
d.is_android_debug_bridge_running()

# 下载文件到本地
d.download_file("/verity_key", "写入到的本地文件")

# 下载文件到 内存/已打开的文件
from io import BytesIO
fd = BytesIO()
d.download_fd("/verity_key", fd)
print (fd.getvalue())

# 注意必须使用 w+b 模式打开被写入文件
fd = open("写入到的本地文件", "wb")
d.download_fd("/verity_key", fd)

# 上传文件到设备
d.upload_file("本地文件路径.txt", "/data/local/tmp/上传到设备上的文件.txt")

# 从 内存/已打开的文件 上传文件
from io import BytesIO
d.upload_fd(BytesIO(b"fileContent"), "/data/local/tmp/上传到设备上的文件.txt")

# 注意必须使用 rb 模式打开文件
fd = open("myfile.txt", "rb")
d.upload_fd(fd, "/data/local/tmp/上传到设备上的文件.txt")

# 删除设备上的文件
d.delete_file("/data/local/tmp/文件.txt")

# 修改设备上的文件权限
d.file_chmod("/data/local/tmp/文件.txt", mode=0o777)

# 获取设备上文件的信息
d.file_stat("/data/local/tmp/文件.txt")

# 关闭系统（等于关机）
d.shutdown()
# 重启系统（等于重启）
d.reboot()

# 关闭设备上运行的 lamda 服务
d.exit()

d.enumerate_all_pkg_names()

d.enumerate_running_processes()

d.current_application()

# 等价于
d.application(d.current_application().applicationId)

# 获取当前前台的 activity
d.current_application().activity

# 由于JSON标准，附加数据暂时只支持 boolean, int 以及 string 类型
d.start_activity(action="some", category="some", component="some",
                 extras={"boolean": False, "int": 1, "string": "me"})

# 例如：启动 设置 APP（当然这几乎等价于直接启动app）
d.start_activity(action="android.intent.action.MAIN",
                 category="android.intent.category.LAUNCHER",
                 component="com.android.settings/.Settings")

# 例如：进入证书设置
d.start_activity(action="com.android.settings.TRUSTED_CREDENTIALS")

app = d.application("com.my.app")

# 获取应用所有权限
app.permissions()
# 授予 READ_PHONE_STATE 权限
app.grant("android.permission.READ_PHONE_STATE", mode=GrantType.GRANT_ALLOW)
# 拒绝 READ_PHONE_STATE 权限
app.grant("android.permission.READ_PHONE_STATE", mode=GrantType.GRANT_DENY)
# 检查是否已授予权限
app.is_permission_granted("android.permission.READ_PHONE_STATE")
# 撤销已授予的权限
app.revoke("android.permission.READ_PHONE_STATE")

# 删除应用的缓存数据
app = d.application("com.my.app")
app.delete_cache()
# 重置应用数据
app.reset_data()

app = d.application("com.my.app")

# 启动应用
app.start()
# 检查应用是否正在前台运行
app.is_foreground()
# 关闭应用
app.stop()

app = d.application("com.my.app")
# 获取应用信息
app.info()

# 检查应用是否已安装
app.is_installed()
# 卸载应用
app.uninstall()

# 查询该应用的启动 Activity（入口活动）
app.query_launch_activity()

# 启用应用
app.enable()
# 禁用应用
app.disable()

wifi = d.stub("Wifi")

# 获取wifi bssid,ssid ip 等相关信息
wifi.status()

# 获取黑名单中的所有 bssid
wifi.blacklist_get_all()

# 将bssid加入黑名单(将不会显示在wifi列表)
wifi.blacklist_add("3c:06:aa:8a:55:66")

# 清空所有黑名单
wifi.blacklist_clear()

# 执行 wifi 扫描
wifi.scan()

# 获取周边 wifi 扫描结果
wifi.scan_results()

# 获取当前wifi的mac地址
wifi.get_mac_addr()

# 获取 wifi 信号强度，链接速率
wifi.signal_poll()

d.device_info()

# 息屏
d.sleep()
# 亮屏
d.wake_up()
# 屏幕是否点亮
d.is_screen_on()
# 屏幕是否已锁定
d.is_screen_locked()

d.set_clipboard("剪切板内容")

# 获取剪切板内容（不支持安卓10+）
d.get_clipboard()

# 此方法可使用如下17种按键
# KEY_BACK
# KEY_CAMERA
# KEY_CENTER
# KEY_DELETE
# KEY_DOWN
# KEY_ENTER
# KEY_HOME
# KEY_LEFT
# KEY_MENU
# KEY_POWER
# KEY_RECENT
# KEY_RIGHT
# KEY_SEARCH
# KEY_UP
# KEY_VOLUME_DOWN
# KEY_VOLUME_MUTE
# KEY_VOLUME_UP
d.press_key(Keys.KEY_BACK)

# 同时为了可以使用更多按键，也可以使用这个方法
d.press_keycode(KeyCodes.KEYCODE_CALL)
# 可使用的 KEYCODE 可以自行查看此文档
# https://developer.android.com/reference/android/view/KeyEvent#KEYCODE_0

quality = 60 # 截图质量，默认为全画质
d.screenshot(quality).save("screenshot.png")
# 截取屏幕上特定区域的图像
# Bound 的参数 top,left 等定义：

# top:     从距离屏幕顶部向下数 top 个像素
# bottom:  从距离屏幕顶部向下数 bottom 个像素
# left:    从距离屏幕左侧向右数 left 个像素
# right:   到距离屏幕左侧向右数 right 个像素

# 正常情况下 top 永远小于 bottom，left 永远小于 right
bound = Bound(top=50, bottom=80, left=50, right=80)
d.screenshot(quality, bound=bound).save("partial.png")

d.click(Point(x=100, y=100))

A = Point(x=100, y=100)
B = Point(x=500, y=500)

d.drag(A, B)

A = Point(x=100, y=100)
B = Point(x=500, y=500)

d.swipe(A, B)

p1 = Point(x=100, y=100)
p2 = Point(x=500, y=500)
p3 = Point(x=200, y=200)

# 从点P1滑动到点P2随后滑动到点P3，可任意个点
d.swipe_points(p1, p2, p3)

d.open_notification()
d.open_quick_settings()

d.dump_window_hierarchy().getvalue()

# 单位是毫秒，5*1000 代表5秒
d.wait_for_idle(5*1000)

# 选择界面上的包含文字 被测APP 的元素
element = d(textContains="被测APP")
# 当然，你不一定要这样赋值到 element，也可直接使用 d(textContains="被测APP")

# 我们现在假设，界面上这个 被测APP 是手机上被测APP的图标名称（图标下面的名称）。
element = d(textContains="被测APP")
# 是否存在该元素
element.exists()
# 点击该元素，不存在则会抛出异常
# Corner.COR_CENTER 代表点击该元素中心点，你可查看 COR_CENTER 定义获取其他可点击的位置
element.click(corner=Corner.COR_CENTER)

# 点击该元素，不存在不会抛出异常
element.click_exists(corner=Corner.COR_CENTER)

# 长按该元素，不存在则会抛出异常
element.long_click(corner=Corner.COR_CENTER)

# 获取元素信息
element.info()

# 获取元素的中心点坐标
element.info().bounds.center()

# 获取元素的左上点坐标
element.info().bounds.corner("top-left")

# 获取元素的高度
element.info().bounds.height

# 获取元素的宽度
element.info().bounds.width

# 获取元素个数
element.count()

# 等待元素出现，最多等待10秒
element.wait_for_exists(10*1000)

# 等待元素消失，最多等待10秒
element.wait_until_gone(10*1000)

# 获取该元素的截图（不是全屏，只是该元素）
# quality 为截图质量 1-100
element.screenshot(quality=60)

# 将此 APP 拖动归类到 购物 文件夹（依据实际情况修改）
element.drag_to(Selector(text="购物"))

#########
# 查找同级或者子级元素
#########
# 有时候会有一些重复元素或者无明显特征的元素，很难去定位
# 这时你可以通过查找子级/同级元素的方法来缩小查找范围
# 子级元素，举例为：一个聊天登录框，里面的输入框即为登录框的子级元素
# 同级元素，举例为：聊天输入框里面的用户名和密码框为同级原始（正常情况下）
form = d(resourceId="login_form")
form.child(index=1)
# 这将获取到 login_form 下 index 为 0 的元素
form.child(index=1).sibling()
# 你也这样来找与 login_form 同级的找回密码按钮
#（其实已经可以通过字符串判断了，就不需要这样做了，这里只是演示）
form.sibling(textContains="找回密码")
# 它们本身就是一个element，你可以对其做任何 element 的操作


############################
# 现在 element 改变了其意义，变为选择了输入框
############################

# 示例为：在一加搜索应用界面的搜索框输入 被测APP

# 注意，不要直接往看似输入框的地方输入文字，可能并无法输入
# 有些输入框需要点击一次才会进入真正的输入框，请使用此真正输入框的资源ID
element = d(resourceId="net.oneplus.launcher:id/search_all_apps")
element.set_text("被测APP")

# 获取输入的内容
element.get_text()

# 清空刚刚输入的内容
element.clear_text_field()

# 配合点击搜索，来完成一次类人的搜索操作。


# 滑动操作（列表上下滑动翻页）
# 注意，这些操作并不保证精度，下面这些方法正常情况下都并不需要选择器，
# 但是你可根据实际情况自行加入选择器

# 向上滑动， step 自行调整，越多会越慢，比较适合精度要求较高的滑动
d().swipe(direction=Direction.DIR_UP, step=32)
# 其他滑动方向：
#DIR_UP     向上滑动
#DIR_LEFT   向左滑动
#DIR_DOWN   向下滑动
#DIR_RIGHT  向右滑动

#########
# fling：甩动，即正常人滑动屏幕的行为，较快
#########
# 从上向下
d().fling_from_top_to_bottom()
# 从下往上
d().fling_from_bottom_to_top()
# 从左往右
d().fling_from_left_to_right()
# 从右往左
d().fling_from_right_to_left()

# 其他，一直向下/左右上滑，直到滑动到底
# 因为并不是一定可以滑动到底或者检测到滑动到底
# 所以 max_swipes 参数是必须的
d().fling_from_top_to_bottom_to_end(max_swipes=32)
d().fling_from_bottom_to_top_to_end(max_swipes=32)
d().fling_from_left_to_right_to_end(max_swipes=32)
d().fling_from_right_to_left_to_end(max_swipes=32)

#########
# scroll: 比较机械性的滑动
#########
step = 60
max_swipes = 32
# 从上往下滑动 step 步
d().scroll_from_top_to_bottom(step)
# 从下往上滑动 step 步
d().scroll_from_bottom_to_top(step)
# 从左往右滑动 step 步
d().scroll_from_left_to_right(step)
# 从右往左滑动 step 步
d().scroll_from_right_to_left(step)

# 其他，一直向下/左右上滑，直到滑动到底
# 同上文 fling 描述
d().scroll_from_top_to_bottom_to_end(max_swipes, step)
d().scroll_from_bottom_to_top_to_end(max_swipes, step)
d().scroll_from_left_to_right_to_end(max_swipes, step)
d().scroll_from_right_to_left_to_end(max_swipes, step)

# 启动监视器循环
d.set_watcher_loop_enabled(True)

# 获取监视器是否已启动
d.get_watcher_loop_enabled()

# 移除系统中应用的所有 watcher，建议每次使用前都执行防止前面任务注册的未删除影响正常处理流程
d.remove_all_watchers()

# 获取系统中所有已应用的 watcher 名称列表
d.get_applied_watchers()

# 彻底移除一个 watcher
d.remove_watcher(name)

# 应用watcher到系统中（当 watcher_loop 启动，此watcher将会生效）
d.set_watcher_enabled(name, True)
# 取消应用
d.set_watcher_enabled(name, False)

# 获取此 watcher 是否应用
d.get_watcher_enabled(name)

# 做一些测试前的清理，当然，并不是每 register 一个就需要这样
# 只是为了确保测试过程不被干扰
d.remove_all_watchers()
d.set_watcher_loop_enabled(True)

# 应用监视界面出现 好的 的次数
# 第二个参数为数组，可以给多个 Selector 表示条件都满足才会记录
# 但是不建议超过三个
d.register_none_op_watcher("RecordElementAppearTimes", [Selector(textContains="好的")])
d.set_watcher_enabled("RecordElementAppearTimes", True)

# ... 做满足条件的操作

# 获取记录的次数
d.get_watcher_triggered_count("RecordElementAppearTimes")

# 重置记录的次数
d.reset_watcher_triggered_count("RecordElementAppearTimes")

# 移除
d.remove_watcher("RecordElementAppearTimes")

# 做一些测试前的清理，当然，并不是每 register 一个就需要这样
# 只是为了确保测试过程不被干扰
d.remove_all_watchers()
d.set_watcher_loop_enabled(True)

# 示例为，当APP启动后出现用户协议时，自动点击同意
# 第二个参数为数组，可以给多个 Selector 表示条件都满足才会点击
# 但是不建议超过三个
d.register_click_target_selector_watcher("ClickAcceptWhenShowAggrement", [Selector(textContains="用户协议")],
                                         Selector(textContains="同意", clickable=True))
d.set_watcher_enabled("ClickAcceptWhenShowAggrement", True)

# ... 做满足条件的操作

# 移除
d.remove_watcher("ClickAcceptWhenShowAggrement")

# 做一些测试前的清理，当然，并不是每 register 一个就需要这样
# 只是为了确保测试过程不被干扰
d.remove_all_watchers()
d.set_watcher_loop_enabled(True)

# 示例为，当界面存在 个人中心 时，按下HOME键回到启动屏幕
# 第二个参数为数组，可以给多个 Selector 表示条件都满足才会点击
# 但是不建议超过三个
d.register_press_key_watcher("PressBackWhenHomePageShows", [Selector(textContains="个人中心")], Keys.KEY_HOME)
d.set_watcher_enabled("PressBackWhenHomePageShows", True)

# ... 做满足条件的操作

# 移除
d.remove_watcher("PressBackWhenHomePageShows")

# 获得锁，此锁将在60秒后被自动释放，其他客户端将可以获得锁，你可以更改此时间
# 但是，如果改得太高因为异常脚本退出，你将近乎永远无法连接设备，你可能需要进行重启
# 此获得锁接口可重入，重入时等价于 _refresh_lock，建议只调用一次
d._acquire_lock(leaseTime=60)
# 刷新锁，每次调用后将锁过期时间设为此 leaseTime
# 做定期调用来保持设备锁定
d._refresh_lock(leaseTime=60)
# 释放锁，其他客户端将可以获得锁
d._release_lock()