github-mirrors/pyPLC
1
0
Fork 0
mirror of https://github.com/uhi22/pyPLC.git synced 2024-11-20 01:13:58 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

introduced address manager

Browse source
This commit is contained in:
uhi22 2022-11-11 11:21:28 +01:00
parent 023e0a7fb5
commit e679123413
6 changed files with 189 additions and 85 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

138
addressManager.py Normal file
View file

@ -0,0 +1,138 @@
import subprocess
import os
from helpers import * # prettyMac etc
MAC_LAPTOP = [0xdc, 0x0e, 0xa1, 0x11, 0x67, 0x08 ] # Win10 laptop
#MAC_RANDOM = [0xaa, 0xbb, 0xcc, 0xdd, 0xee, 0xff ]
#MAC_ALPI = [0x0A, 0x19, 0x4A, 0x39, 0xD6, 0x98 ] # alpitronics
#MAC_TPLINK_E4 = [0x98, 0x48, 0x27, 0x5A, 0x3C, 0xE4 ] # TPlink PLC adaptor
#MAC_TPLINK_E6 = [0x98, 0x48, 0x27, 0x5A, 0x3C, 0xE6 ] # TPlink PLC adaptor
#MAC_DEVOLO_26 = [0xBC, 0xF2, 0xAF, 0x0B, 0x8E, 0x26 ] # Devolo PLC adaptor
class addressManager():
def __init__(self):
self.localIpv6Addresses=[]
self.findLocalMacAddress()
self.findLinkLocalIpv6Address()
pass
def findLinkLocalIpv6Address(self):
# For the IPv4 address, this works:
# print(socket.gethostbyname(socket.gethostname())) # Result: The ipv4 address of the ethernet adaptor
# But this does not help, we need the IPv6 address.
#
# On windows we can use command line tool "ipconfig".
# The link-local addresses always start with fe80::, so we just parse the output of ipconfig line-by-line.
# If we have multiple interfaces (e.g. ethernet and WLAN), it will find multiple link-local-addresses.
foundAddresses = []
if os.name == 'nt':
result = subprocess.run(["ipconfig.exe"], capture_output=True, text=True, encoding="ansi")
if (len(result.stderr)>0):
print(result.stderr)
else:
lines = result.stdout.split("\n")
for line in lines:
if (line.find("IPv6")>0):
k = line.find(" fe80::")
if (k>0):
foundAddresses.append(line[k+1:])
print("[addressManager] Found " + str(len(foundAddresses)) + " link-local IPv6 addresses.")
for a in foundAddresses:
print(a)
self.localIpv6Addresses = foundAddresses
if (len(self.localIpv6Addresses)==0):
print("[addressManager] Error: No local Ipv6 address was found.")
self.localIpv6Address = "localhost"
else:
# at least one address was found. Take the first one (this may be the wrong adaptor).
self.localIpv6Address = self.localIpv6Addresses[0]
if (len(self.localIpv6Addresses)>1):
print("[addressManager] Warning: Multiple Ipv6 addresses were found. If you have multiple network adaptors, try to disconnect the not relevant ones.")
print("[addressManager] Using the first detected, " + self.localIpv6Addresses[0])
print("[addressManager] Local IPv6 is " + self.localIpv6Address)
def findLocalMacAddress(self):
# Find out the MAC address of the local ethernet interface.
# Todo: Find this out dynamically.
self.localMac = MAC_LAPTOP
print("[addressManager] we have local MAC " + prettyMac(self.localMac) + ". Todo: find this out dynamically.")
def setPevMac(self, pevMac):
# During the SLAC, the MAC of the PEV was found out. Store it, maybe we need it later.
self.pevMac = pevMac
print("[addressManager] pev has MAC " + prettyMac(self.pevMac))
def setPevIp(self, pevIp):
# During SDP, the IPv6 of the PEV was found out. Store it, maybe we need it later.
if (type(pevIp)==type(bytearray([0]))):
# the parameter was a bytearray. We want a string, so convert it.
# print("this is a byte array")
if (len(pevIp)!=16):
print("[addressManager] ERROR: setPevIp: invalid ip address len " + str(len(pevIp)))
return
s = ""
for i in range(0, 16):
s = s + twoCharHex(pevIp[i])
if (((i % 2)==1) and (i!=15)):
s = s + ":"
self.pevIp = s.lower()
else:
# the parameter was a string, assumingly. Just take it over.
self.pevIp = pevIp
print("[addressManager] pev has IP " + self.pevIp)
def getLocalMacAddress(self):
print("[addressManager] will give local MAC " + prettyMac(self.localMac))
return self.localMac;
def getLinkLocalIpv6Address(self, resulttype="string"):
if (resulttype=="string"):
return self.localIpv6Address;
if (resulttype=="bytearray"):
# The caller wants a byte array. We need to convert a string like
# "fe80::4c46:fea5:b6c9:25a9%3" into a byte array of 16 bytes.
# "fe80::4c46:fea5:b6c9:25a9"
ba = bytearray(16)
s = self.localIpv6Address
print("[addressManager] converting self.localIpv6Address into bytearray")
# Step1: remove the % and all behind:
x = s.find("%")
#print("percent found at " + str(x))
#print(s)
if (x>0):
s=s[0:x]
print(s)
# Step 2: expand the ommited zeros
x = s.find("::")
#print(":: found at " + str(x))
if (x>0):
# a :: means four bytes which are 0x00 each.
# Todo: but we need two bytes more?!?
s = s.replace("::", ":0000:0000:0000:")
print(s)
# Step 3: Remove all ":"
s = s.replace(":", "")
print(s)
if (len(s)!=32):
print("[addressManager] ERROR: invalid length of IPv6 string. Expected be 16 bytes, means 32 hex characters. Found " + str(len(s)))
else:
for i in range(0, 16):
sTwoChar = s[2*i : 2*i+2]
ba[i] = int(sTwoChar, 16)
#print(sTwoChar)
#fe80::4c46:fea5:b6c9:25a9%3
return ba
if __name__ == "__main__":
print("Testing addressManager.py....")
am = addressManager()
#am.setPevIp(bytearray([0xfe, 0x80]))
am.setPevIp(bytearray([0xfe, 0x80, 0x00, 0x00, 0xfe, 0x80, 0x00, 0x00, 0xfe, 0x80, 0x00, 0x00, 0xDE, 0xAD, 0xBE, 0xEF]))
print(am.getLinkLocalIpv6Address())
print(am.getLinkLocalIpv6Address(resulttype="bytearray"))

7
helpers.py
View file

@ -20,7 +20,10 @@ def prettyHexMessage(mybytearray, description=""):
def prettyMac(macByteArray):
s=""
for i in range(0, 5):
length = len(macByteArray)
if (length!=6):
s="invalid MAC length " + str(length) + "!"
for i in range(0, length-1):
s = s + twoCharHex(macByteArray[i]) + ":"
s = s + twoCharHex(macByteArray[i])
s = s + twoCharHex(macByteArray[length-1])
return s

55
pyPlcHomeplug.py
View file

@ -39,12 +39,6 @@ from helpers import * # prettyMac etc
from pyPlcModes import *
MAC_BROADCAST = [0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF ]
MAC_LAPTOP = [0xdc, 0x0e, 0xa1, 0x11, 0x67, 0x08 ] # Win10 laptop
MAC_RANDOM = [0xaa, 0xbb, 0xcc, 0xdd, 0xee, 0xff ]
MAC_ALPI = [0x0A, 0x19, 0x4A, 0x39, 0xD6, 0x98 ] # alpitronics
MAC_TPLINK_E4 = [0x98, 0x48, 0x27, 0x5A, 0x3C, 0xE4 ] # TPlink PLC adaptor
MAC_TPLINK_E6 = [0x98, 0x48, 0x27, 0x5A, 0x3C, 0xE6 ] # TPlink PLC adaptor
MAC_DEVOLO_26 = [0xBC, 0xF2, 0xAF, 0x0B, 0x8E, 0x26 ] # Devolo PLC adaptor
CM_SET_KEY = 0x6008
@ -167,7 +161,6 @@ class pyPlcHomeplug():
#self.fillDestinationMac(MAC_TPLINK_E4)
self.fillDestinationMac(MAC_BROADCAST)
# Source MAC
#self.fillSourceMac(MAC_LAPTOP)
self.fillSourceMac(self.myMAC)
# Protocol
self.mytransmitbuffer[12]=0x88 # Protocol HomeplugAV
@ -216,7 +209,6 @@ class pyPlcHomeplug():
#self.fillDestinationMac(MAC_TPLINK_E4)
self.fillDestinationMac(MAC_BROADCAST)
# Source MAC
#self.fillSourceMac(MAC_LAPTOP)
self.fillSourceMac(self.myMAC)
# Protocol
self.mytransmitbuffer[12]=0x88 # Protocol HomeplugAV
@ -363,43 +355,6 @@ class pyPlcHomeplug():
# 92 reserved 0
self.setNmkAt(93) # 93 to 108 NMK. We can freely choose this. Normally we should use a random number.
def composeDHCP(self):
# DHCP discover, to check whether this "normal" package arrives on the other side
self.mytransmitbuffer = bytearray(379)
self.cleanTransmitBuffer()
# Destination MAC
self.fillDestinationMac(MAC_BROADCAST)
# Source MAC
self.fillSourceMac(MAC_LAPTOP)
self.mytransmitbuffer[12]=0x08 # Protocol IPv4
self.mytransmitbuffer[13]=0x00
self.mytransmitbuffer[14]=0x45 # header len
self.mytransmitbuffer[15]=0x00 #
self.mytransmitbuffer[16]=0x01 #
self.mytransmitbuffer[17]=0x6D #
self.mytransmitbuffer[18]=0x9B #
self.mytransmitbuffer[19]=0x30 #
self.mytransmitbuffer[20]=0x00 #
self.mytransmitbuffer[21]=0x00 #
self.mytransmitbuffer[22]=0x40 #
self.mytransmitbuffer[23]=0x11 #
self.mytransmitbuffer[24]=0xde #
self.mytransmitbuffer[25]=0x50 #
self.mytransmitbuffer[26]=0x00 #
self.mytransmitbuffer[27]=0x00 #
self.mytransmitbuffer[28]=0x00 #
self.mytransmitbuffer[29]=0x00 #
self.mytransmitbuffer[30]=0xFF #
self.mytransmitbuffer[31]=0xFF #
self.mytransmitbuffer[32]=0xFF #
self.mytransmitbuffer[33]=0xFF #
self.mytransmitbuffer[34]=0x00 #
self.mytransmitbuffer[35]=0x44 #
self.mytransmitbuffer[36]=0x00 #
self.mytransmitbuffer[37]=0x43 #
def sendTestFrame(self, selection):
@ -456,6 +411,7 @@ class pyPlcHomeplug():
for i in range(0, 6):
self.pevMac[i] = self.myreceivebuffer[6+i]
self.addressManager.setPevMac(self.pevMac)
self.showStatus(prettyMac(self.pevMac), "pevmac")
# If we want to emulate an EVSE, we want to answer.
if (self.iAmEvse==1):
@ -551,11 +507,12 @@ class pyPlcHomeplug():
self.ipv6.enterListenMode()
self.showStatus("LISTEN mode", "mode")
def __init__(self, callbackAddToTrace=None, callbackShowStatus=None, mode=C_LISTEN_MODE):
def __init__(self, callbackAddToTrace=None, callbackShowStatus=None, mode=C_LISTEN_MODE, addrMan=None):
self.mytransmitbuffer = bytearray("Hallo das ist ein Test", 'UTF-8')
self.nPacketsReceived = 0
self.callbackAddToTrace = callbackAddToTrace
self.callbackShowStatus = callbackShowStatus
self.addressManager = addrMan
#self.sniffer = pcap.pcap(name=None, promisc=True, immediate=True, timeout_ms=50)
# eth3 means: Third entry from back, in the list of interfaces, which is provided by pcap.findalldevs.
# Improvement necessary: select the interface based on the name.
@ -576,10 +533,10 @@ class pyPlcHomeplug():
self.sniffer.setnonblock(True)
self.NMK = [ 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16 ] # a default network key
self.NID = [ 1, 2, 3, 4, 5, 6, 7 ] # a default network ID
self.pevMac = [0x55, 0x56, 0x57, 0x58, 0x59, 0x5A ] # a default pev MAC
self.myMAC = MAC_LAPTOP # todo: find this dynamically
self.pevMac = [0x55, 0x56, 0x57, 0x58, 0x59, 0x5A ] # a default pev MAC. Will be overwritten later.
self.myMAC = self.addressManager.getLocalMacAddress()
self.runningCounter=0
self.ipv6 = pyPlcIpv6.ipv6handler(self.transmit)
self.ipv6 = pyPlcIpv6.ipv6handler(self.transmit, self.addressManager)
self.ipv6.ownMac = self.myMAC
if (mode == C_LISTEN_MODE):
self.enterListenMode()

32
pyPlcIpv6.py
View file

@ -16,7 +16,7 @@
# SLAAC: Stateless auto address configuration (not SLAC!). A method to automatically set IPv6 address, based
# on the 6 byte MAC address.
from helpers import showAsHex, prettyHexMessage
from helpers import showAsHex, prettyHexMessage, prettyMac
import udpChecksum
@ -63,7 +63,7 @@ class ipv6handler():
self.IpResponse[24+i] = self.EvccIp[i] # destination IP address
for i in range(0, len(buffer)):
self.IpResponse[40+i] = buffer[i]
showAsHex(self.IpResponse, "IP response ")
#showAsHex(self.IpResponse, "IP response ")
self.packResponseIntoEthernet(self.IpResponse)
@ -86,7 +86,7 @@ class ipv6handler():
self.UdpResponse[7] = 0
for i in range(0, len(buffer)):
self.UdpResponse[8+i] = buffer[i]
showAsHex(self.UdpResponse, "UDP response ")
#showAsHex(self.UdpResponse, "UDP response ")
# The content of buffer is ready. We can calculate the checksum. see https://en.wikipedia.org/wiki/User_Datagram_Protocol
checksum = udpChecksum.calculateUdpChecksumForIPv6(self.UdpResponse, self.SeccIp, self.EvccIp)
self.UdpResponse[6] = checksum >> 8
@ -128,6 +128,10 @@ class ipv6handler():
if (self.destinationport == 15118): # port for the SECC
if ((self.udpPayload[0]==0x01) and (self.udpPayload[1]==0xFE)): # protocol version 1 and inverted
# it is a V2GTP message
if (self.iAmEvse):
# if we are the charger, lets save the cars IP for later use.
self.EvccIp = self.sourceIp
self.addressManager.setPevIp(self.EvccIp)
showAsHex(self.udpPayload, "V2GTP ")
self.evccPort = self.sourceport
v2gptPayloadType = self.udpPayload[2] * 256 + self.udpPayload[3]
@ -194,6 +198,10 @@ class ipv6handler():
# The evaluation function for received ipv6 packages.
if (len(pkt)>60):
self.myreceivebuffer = pkt
# extract the source ipv6 address
self.sourceIp = bytearray(16)
for i in range(0, 16):
self.sourceIp[i] = self.myreceivebuffer[22+i]
self.nextheader = self.myreceivebuffer[20]
if (self.nextheader == 0x11): # it is an UDP frame
self.sourceport = self.myreceivebuffer[54] * 256 + self.myreceivebuffer[55]
@ -212,15 +220,23 @@ class ipv6handler():
if (self.nextheader == 0x06): # it is an TCP frame
self.evaluateTcpPacket()
def __init__(self, transmitCallback):
#self.enterEvseMode()
self.enterListenMode()
def __init__(self, transmitCallback, addressManager):
self.enterEvseMode()
#self.enterListenMode()
self.transmit = transmitCallback
self.addressManager = addressManager
# 16 bytes, a default IPv6 address for the charging station
# self.SeccIp = [ 0xfe, 0x80, 0, 0, 0, 0, 0, 0, 0x06, 0xaa, 0xaa, 0xff, 0xfe, 0, 0xaa, 0xaa ]
# fe80::e0ad:99ac:52eb:85d3 is the Win10 laptop
self.SeccIp = [ 0xfe, 0x80, 0, 0, 0, 0, 0, 0, 0xe0, 0xad, 0x99, 0xac, 0x52, 0xeb, 0x85, 0xd3 ]
# self.SeccIp = [ 0xfe, 0x80, 0, 0, 0, 0, 0, 0, 0xe0, 0xad, 0x99, 0xac, 0x52, 0xeb, 0x85, 0xd3 ]
# 16 bytes, a default IPv6 address for the vehicle
# todo: On EVSE side, extract the vehicles IP address from the SDP communication
self.EvccIp = [ 0xfe, 0x80, 0, 0, 0, 0, 0, 0, 0x06, 0x65, 0x65, 0xff, 0xfe, 0, 0x64, 0xC3 ]
self.ownMac = [ 0x01, 0x02, 0x03, 0x04, 0x05, 0x06 ] # 6 bytes own MAC default. Should be overwritten before use.
#self.ownMac = [ 0x01, 0x02, 0x03, 0x04, 0x05, 0x06 ] # 6 bytes own MAC default. Should be overwritten before use.
self.ownMac = self.addressManager.getLocalMacAddress()
print("pyPlcIpv6 started with ownMac " + prettyMac(self.ownMac))
if (self.iAmEvse):
# If we are an charger, we need to support the SDP, which requires to know our IPv6 adrress.
self.SeccIp = self.addressManager.getLinkLocalIpv6Address("bytearray")

34
pyPlcTcpSocket.py
View file

@ -108,6 +108,15 @@ class pyPlcTcpServerSocket():
self.ourSocket.bind((self.ipAdress, self.tcpPort))
self.ourSocket.listen(1)
print("pyPlcTcpSocket listening on port " + str(self.tcpPort))
hostname=socket.gethostname()
IPAddr=socket.gethostbyname(hostname)
addressInfo = socket.getaddrinfo(hostname, None, socket.AF_INET6)
#print("Your Computer Name is:"+hostname)
print("The socket is linked the following IP addresses:")
for i in range(0, len(addressInfo)):
#fe80::4c46:fea5:b6c9:25a9
IPv6Addr = addressInfo[i][4][0]
print(IPv6Addr)
self.read_list = [self.ourSocket]
self.rxData = []
@ -214,31 +223,8 @@ def testClientSocket():
def testExtra():
print("testExtra")
findLinkLocalIpv6Address()
#findLinkLocalIpv6Address()
def findLinkLocalIpv6Address():
# For the IPv4 address, this works:
# print(socket.gethostbyname(socket.gethostname())) # Result: The ipv4 address of the ethernet adaptor
# But this does not help, we need the IPv6 address.
#
# On windows we can use command line tool "ipconfig".
# The link-local addresses always start with fe80::, so we just parse the output of ipconfig line-by-line.
# If we have multiple interfaces (e.g. ethernet and WLAN), it will find multiple link-local-addresses.
foundAddresses = []
if os.name == 'nt':
result = subprocess.run(["ipconfig.exe"], capture_output=True, text=True, encoding="ansi")
if (len(result.stderr)>0):
print(result.stderr)
else:
lines = result.stdout.split("\n")
for line in lines:
if (line.find("IPv6")>0):
k = line.find(" fe80::")
if (k>0):
foundAddresses.append(line[k+1:])
print("Found " + str(len(foundAddresses)) + " link-local addresses.")
for a in foundAddresses:
print(a)
if __name__ == "__main__":
print("Testing pyPlcTcpSocket.py....")

6
pyPlcWorker.py
View file

@ -8,6 +8,8 @@ import pyPlcHomeplug
import fsmEvse
import fsmPev
from pyPlcModes import *
import addressManager
class pyPlcWorker():
def __init__(self, callbackAddToTrace=None, callbackShowStatus=None, mode=C_EVSE_MODE):
@ -15,9 +17,11 @@ class pyPlcWorker():
self.nMainFunctionCalls=0
self.mode = mode
self.strUserAction = ""
self.addressManager = addressManager.addressManager()
self.addressManager.findLinkLocalIpv6Address()
self.callbackAddToTrace = callbackAddToTrace
self.callbackShowStatus = callbackShowStatus
self.hp = pyPlcHomeplug.pyPlcHomeplug(self.callbackAddToTrace, self.callbackShowStatus, self.mode)
self.hp = pyPlcHomeplug.pyPlcHomeplug(self.callbackAddToTrace, self.callbackShowStatus, self.mode, self.addressManager)
if (self.mode == C_EVSE_MODE):
self.evse = fsmEvse.fsmEvse()
if (self.mode == C_PEV_MODE):