Ruby class for accessing SENEC solar system data via their app API

senec_oauth2.rb

# This Gist is only a proof of concept and will never be updated.
# The latest implementation can be found in my Ruby Gem:
# https://github.com/solectrus/senec/blob/main/lib/senec/cloud/connection.rb

require 'oauth2'

class SenecOauth2
  # Configuration constants derived from publicly available sources on GitHub

  CONFIG_URL =
    'https://sso.senec.com/realms/senec/.well-known/openid-configuration'

  CLIENT_ID = 'endcustomer-app-frontend'
  REDIRECT_URI = 'senec-app-auth://keycloak.prod'
  SCOPE = 'roles meinsenec'

  SYSTEMS_HOST = 'https://senec-app-systems-proxy.prod.senec.dev'
  MEASUREMENTS_HOST = 'https://senec-app-measurements-proxy.prod.senec.dev'
  WALLBOX_HOST = 'https://senec-app-wallbox-proxy.prod.senec.dev'

  def initialize(username, password)
    @username = username
    @password = password
  end

  attr_reader :username, :password

  def systems
    fetch_payload "#{SYSTEMS_HOST}/v1/systems"
  end

  def system_details(system_id)
    fetch_payload "#{SYSTEMS_HOST}/systems/#{system_id}/details"
  end

  def dashboard(system_id)
    fetch_payload "#{MEASUREMENTS_HOST}/v1/systems/#{system_id}/dashboard"
  end

  def wallbox(system_id, wallbox_id)
    fetch_payload "#{WALLBOX_HOST}/v1/systems/#{system_id}/wallboxes/#{wallbox_id}"
  end

  private

  attr_accessor :oauth_token

  def authenticate!
    code_verifier = SecureRandom.alphanumeric(43)
    code_challenge = generate_code_challenge(code_verifier)

    auth_url =
      oauth_client.auth_code.authorize_url(
        redirect_uri: REDIRECT_URI,
        scope: SCOPE,
        code_challenge:,
        code_challenge_method: 'S256',
      )

    # Manual HTTP needed for Keycloak cross-domain form handling
    login_form_url = fetch_login_form_url(auth_url)
    redirect_url = submit_credentials(login_form_url)
    authorization_code = extract_authorization_code(redirect_url)

    self.oauth_token =
      oauth_client.auth_code.get_token(
        authorization_code,
        redirect_uri: REDIRECT_URI,
        code_verifier:,
      )
  end

  def generate_code_challenge(code_verifier)
    digest = Digest::SHA256.digest(code_verifier)
    Base64.urlsafe_encode64(digest).delete('=')
  end

  def fetch_login_form_url(auth_url)
    response = http_request(:get, auth_url)
    store_cookies(response) # Required for Keycloak CSRF protection
    extract_form_action_url(response.body)
  end

  def extract_form_action_url(html)
    forms = html.scan(%r{<form[^>]*action="([^"]+)"[^>]*>(.*?)</form>}mi)

    forms.each do |action_url, form_content|
      has_username = form_content.match(/name=["']?username["']?/i)
      has_password = form_content.match(/name=["']?password["']?/i)

      return action_url if has_username && has_password
    end

    raise 'Login form not found'
  end

  def submit_credentials(form_url)
    credentials = { username:, password: }
    response = http_request(:post, form_url, data: credentials)
    raise 'Login failed' unless response.is_a?(Net::HTTPFound)

    response['Location'] || raise('No redirect location')
  end

  def extract_authorization_code(redirect_url)
    raise 'Invalid redirect URL' unless redirect_url&.start_with?(REDIRECT_URI)

    code = CGI.parse(URI(redirect_url).query).dig('code', 0)
    raise 'No authorization code found' unless code

    code
  end

  def ensure_token_valid
    return false unless oauth_token
    return true unless oauth_token.expired?

    puts 'Token expired, refreshing...'
    self.oauth_token = oauth_token.refresh!
    true
  rescue OAuth2::Error => e
    warn "Token refresh failed: #{e.message}"
    false
  end

  def fetch_payload(url, default = nil)
    authenticate! unless oauth_token

    return default unless ensure_token_valid

    response = oauth_token.get(url)
    return default unless response.status == 200

    JSON.parse(response.body)
  rescue => e
    warn "API error: #{e.message}"
    default
  end

  # OAuth2 gem can't handle cross-domain forms with cookies
  def http_request(method, url, data: nil)
    uri = URI(url)
    http = Net::HTTP.new(uri.host, uri.port)
    http.use_ssl = true

    request =
      method == :get ? Net::HTTP::Get.new(uri) : Net::HTTP::Post.new(uri)
    apply_cookies(request)

    request.body =
      if data.is_a?(Hash)
        URI.encode_www_form(data)
      elsif data.is_a?(String)
        data
      end

    http.request(request)
  end

  def apply_cookies(request)
    return if cookies.empty?

    request['Cookie'] = cookies.map { |k, v| "#{k}=#{v}" }.join('; ')
  end

  def store_cookies(response)
    Array(response.get_fields('Set-Cookie')).each do |cookie|
      name, value = cookie.split(';').first.split('=', 2)
      cookies[name] = value if name && value
    end
  end

  def oauth_client
    @oauth_client ||=
      OAuth2::Client.new(
        CLIENT_ID,
        nil,
        site: openid_config['issuer'],
        authorize_url: openid_config['authorization_endpoint'],
        token_url: openid_config['token_endpoint'],
      )
  end

  # Load OpenID configuration from the Keycloak server
  def openid_config
    @openid_config ||= JSON.parse(Net::HTTP.get(URI(CONFIG_URL)))
  rescue => e
    raise "Failed to load OpenID configuration: #{e.message}"
  end

  def cookies
    @cookies ||= {}
  end
end