0. 我們如何通過郵件系統完成遠程控制電腦(關機、重啟等)?
實現思路:
需要有兩個郵箱:接收指令郵箱(A)發送指令郵箱(B)
被控制的電腦(查看 A 郵箱):
1. 每隔指定時間監聽 A 郵箱 查看最近的郵件
2. 打開最近郵件,看是否是 B 郵箱地址發來的執行命令
3. 向 A 和 B 郵箱都發送 主題為:反饋 內容為:“目標計算機已收到指令!開始執行:xxx 指令!” 郵件
4. 執行指定郵箱發送的執行命令
遠程人員操作:
登錄 B 郵箱,向 A 郵箱發送 主題為:“目標計算機執行” 內容為:xxx指令(例如:關機)
1. 代碼編寫
import poplibimport smtplibimport osimport timefrom email.parser import Parserfrom email.header import decode_headerfrom email.utils import parseaddrfrom email.mime.text import MIMETextfrom email.header import Header email_params = dict()email_params['email_message'] = '反饋'email_params['email_subject'] = '目標計算機反饋'email_params['email_from'] = 'A郵箱'email_params['email_user'] = 'A郵箱'email_params['email_exec'] = 'B郵箱'email_params['email_to'] = ['A郵箱', 'B郵箱']email_params['email_smtp'] = 'smtp.163.com'email_params['email_pop'] = 'pop.163.com'email_params['email_pass'] = 'A郵箱的密碼(smtp、pop3的密碼)'interval_time = 5 def decode_str(s): value, charset = decode_header(s)[0] if charset: value = value.decode(charset) return value def get_headers(message): headers = dict() for header in ['From', 'To', 'Subject']: value = message.get(header, '') if value: if header == 'Subject': headers['Subject'] = decode_str(value) elif header == 'From': hdr, addr = parseaddr(value) headers['From'] = decode_str(addr) elif header == 'To': hdr, addr = parseaddr(value) headers['To'] = decode_str(addr) return headers def guess_charset(message): charset = message.get_charset() if charset is None: content_type = message.get('Content-Type', '').lower() pos = content_type.find('charset=') if pos >= 0: charset = content_type[pos + 8:].strip() return charset def get_content(message): for part in message.walk(): content_type = part.get_content_type() charset = guess_charset(part) if content_type == 'text/plain' and charset: try: return part.get_payload(decode=True).decode(charset) except AttributeError: print('type error') except LookupError: print("unknown encoding: utf-8") def receive_email(): email_server = poplib.POP3_SSL(email_params['email_pop']) email_server.user(email_params['email_user']) email_server.pass_(email_params['email_pass']) resp, mails, octets = email_server.list() index = len(mails) if index == 0: return None, None resp, lines, octets = email_server.retr(index) message = b'/r/n'.join(lines).decode('utf-8', 'ignore') message = Parser().parsestr(message) email_server.quit() return get_headers(message), get_content(message) def send_email(email_message): message = MIMEText(email_message, 'plain', 'utf-8') message['Subject'] = Header(email_params['email_subject'], 'utf-8') message['From'] = Header(email_params['email_from']) message['To'] = Header(','.join(email_params['email_to']), 'utf-8') email_server = smtplib.SMTP_SSL(email_params['email_smtp']) email_server.login(email_params['email_from'], email_params['email_pass']) email_server.sendmail(email_params['email_from'], email_params['email_to'], message.as_string()) email_server.close() def exec_operator(exec_content): if exec_content == '關機': os.system("shutdown -s -t 1") elif exec_content == '重啟': os.system("shutdown -r") else: os.system(exec_content) if __name__ == '__main__': while True: headers, content = receive_email() if headers and content and headers['From'] == email_params['email_exec'] and headers['Subject'] == '目標計算機執行': email_message = "目標計算機已接收到指令!開始執行: " + content + " 指令!" send_email(email_message) exec_operator(content) time.sleep(interval_time)
|
新聞熱點
疑難解答