8000 Get Token - 2878 · Issue #117 · atxbyea/samsungrac · GitHub
[go: up one dir, main page]
More Web Proxy on the site http://driver.im/
Skip to content

Get Token - 2878 #117

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the communi E743 ty.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
GSzabados opened this issue Nov 14, 2024 · 2 comments
Open

Get Token - 2878 #117

GSzabados opened this issue Nov 14, 2024 · 2 comments

Comments

@GSzabados
Copy link

Hi,

I have mocked up a script (with the use of ChatGPT) to get the token from an old version (2878) AC.

It can be used from HA terminal like python3 .\ac_2878_get_token.py 192.168.1.100 , just need to include in the same folder the ac14k_m.pem file.

Could you please include it as a file and add some reference to the readme? If you accept this kind of PRs, I will make one then.

import socket
import ssl
import logging
import re
import argparse

class SamsungAirconditioner:
    def __init__(self, ip, logger):
        self.options = {'ip': ip}
        self.logger = logger
        self.token = None

    def get_token(self, callback):
        if not callable(callback):
            raise ValueError("callback is mandatory for get_token")
        # Setup SSL context with TLSv1 and AES256-SHA cipher
        context = ssl.SSLContext(ssl.PROTOCOL_TLSv1)  # Force TLSv1
        context.set_ciphers('ALL:@SECLEVEL=0')
        
        # Load the certificate you trust (cert.pem)
        context.load_cert_chain(certfile='ac14k_m.pem')
        context.check_hostname = False  # Disable hostname verification for self-signed certs
        context.verify_mode = ssl.CERT_NONE  # Optional certificate verification (allow self-signed)

        try:
            with socket.create_connection((self.options['ip'], 2878)) as raw_socket:
                with context.wrap_socket(raw_socket, server_hostname=self.options['ip']) as tls_socket:
                    self.logger.info('connected', {'ipaddr': self.options['ip'], 'port': 2878, 'tls': True})

                    # Receive loop to handle messages line-by-line
                    buffer = ""
                    while True:
                        data = tls_socket.recv(1024).decode('utf-8')
                        if not data:
                            if not self.token:
                                callback(Exception('premature eof'))
                            break

                        buffer += data
                        while '\n' in buffer:
                            line, buffer = buffer.split('\n', 1)
                            self.logger.debug('read: %s', line.strip())

                            # Handle specific responses
                            if line.strip() == 'DRC-1.00':
                                continue
                            if line.strip() == '<?xml version="1.0" encoding="utf-8" ?><Update Type="InvalidateAccount"/>':
                                tls_socket.sendall(b'<Request Type="GetToken" />\r\n')
                                continue
                            if line.strip() == '<?xml version="1.0" encoding="utf-8" ?><Response Type="GetToken" Status="Ready"/>':
                                self.logger.info('waiting for token - turn on the AC in the next 30 seconds')
                                continue
                            if line.strip() == '<?xml version="1.0" encoding="utf-8" ?><Response Status="Fail" Type="Authenticate" ErrorCode="301" />':
                                callback(Exception('Failed authentication'))
                                return

                            # Check for token in response
                            token_match = re.search(r'Token="(.*?)"', line)
                            if token_match:
                                self.token = token_match.group(1)
                                self.logger.info('authenticated')
                                callback(None, self.token)
                                return

                            # Handle status updates
                            if 'Update Type="Status"' in line:
                                state_match = re.search(r'Attr ID="(.*?)" Value="(.*?)"', line)
                                if state_match:
                                    state = {state_match.group(1): state_match.group(2)}
                                    self.logger.info('stateChange', state)
                                    continue

                            # Handle device state
                            if 'Response Type="DeviceState" Status="Okay"' in line:
                                state = {}
                                attributes = line.split("><")
                                for attr in attributes:
                                    attr_match = re.search(r'Attr ID="(.*?)" Type=".*?" Value="(.*?)"', attr)
                                    if attr_match:
                                        state[attr_match.group(1)] = attr_match.group(2)
                                self.logger.info('stateChange', state)

        except (socket.error, ssl.SSLError) as e:
            if not self.token:
                callback(e)


# Set up a basic logger
logger = logging.getLogger('SamsungAirconditioner')
logger.setLevel(logging.DEBUG)
handler = logging.StreamHandler()
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
handler.setFormatter(formatter)
logger.addHandler(handler)

# Define the callback function to handle the response from get_token
def token_callback(error, token=None):
    if error:
        print("Error: {}".format(error))
    else:
        print("Retrieved token: {}".format(token))

# Set up argument parser for command-line input
parser = argparse.ArgumentParser(description="Samsung Air Conditioner Token Retrieval")
parser.add_argument('ip', type=str, help='IP address of the Samsung Air Conditioner')
args = parser.parse_args()

# Create an instance of the SamsungAirconditioner class with the provided IP
aircon = SamsungAirconditioner(ip=args.ip, logger=logger)

# Call the get_token method, passing the token_callback function
aircon.get_token(token_callback)
@mrknowitall526
Copy link

Thanks so much!!! This worked! After spending 4 days trying to do this, this was so easy!

@GSzabados
Copy link
Author

@mrknowitall526, you are very welcome. I am happy that someone has found it useful already.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Labels
None yet
Projects
None yet
Development

No branches or pull requests

2 participants
0