diff --git a/backend/services/acs/.env b/backend/services/acs/.env new file mode 100644 index 0000000..e69de29 diff --git a/backend/services/acs/.gitignore b/backend/services/acs/.gitignore new file mode 100644 index 0000000..e45d6e3 --- /dev/null +++ b/backend/services/acs/.gitignore @@ -0,0 +1 @@ +*.local \ No newline at end of file diff --git a/backend/services/acs/README.md b/backend/services/acs/README.md new file mode 100644 index 0000000..d5cb5d6 --- /dev/null +++ b/backend/services/acs/README.md @@ -0,0 +1,97 @@ +# Moses ACS [![Build Status](https://travis-ci.org/lucacervasio/mosesacs.svg?branch=master)](https://travis-ci.org/lucacervasio/mosesacs) + +An ACS in Go for provisioning CPEs, suitable for test purposes or production deployment. + +## Getting started + +Install the package: + + go get oktopUSP/backend/services/acs + +Run daemon: + + mosesacs -d + +Connect to it and get a cli: + + mosesacs + +Congratulations, you've connected to the daemon via websocket. Now you can issue commands via CLI or browse the embedded webserver at http://localhost:9292/www + +## Compatibility on ARM + +Moses is built on purpose only with dependencies in pure GO. So it runs on ARM processors with no issues. We tested it on QNAP devices and Raspberry for remote control. + +## CLI commands + +### 1. `list`: list CPEs + + example: + +``` + moses@localhost:9292/> list + cpe list + CPE A54FD with OUI 006754 +``` + +### 2. `readMib SERIAL LEAF/SUBTREE`: read a specific leaf or a subtree + + example: + +``` + moses@localhost:9292/> readMib A54FD Device. + Received an Inform from [::1]:58582 (3191 bytes) with SerialNumber A54FD and EventCodes 6 CONNECTION REQUEST + InternetGatewayDevice.Time.NTPServer1 : pool.ntp.org + InternetGatewayDevice.Time.CurrentLocalTime : 2014-07-11T09:08:25 + InternetGatewayDevice.Time.LocalTimeZone : +00:00 + InternetGatewayDevice.Time.LocalTimeZoneName : Greenwich Mean Time : Dublin + InternetGatewayDevice.Time.DaylightSavingsUsed : 0 +``` + +### 3. `writeMib SERIAL LEAF VALUE`: issue a SetParameterValues and write a value into a leaf + + example: + +``` + moses@localhost:9292/> writeMib A54FD InternetGatewayDevice.Time.Enable false + Received an Inform from [::1]:58582 (3191 bytes) with SerialNumber A54FD and EventCodes 6 CONNECTION REQUEST +``` + +### 4. `GetParameterNames SERIAL LEAF/SUBTREE`: issue a GetParameterNames and get all leaves/objects at first level + + example: + +``` +moses@localhost:9292/> GetParameterNames A54FD InternetGatewayDevice. +Received an Inform from [::1]:55385 (3119 bytes) with SerialNumber A54FD and EventCodes 6 CONNECTION REQUEST +InternetGatewayDevice.LANDeviceNumberOfEntries : 0 +InternetGatewayDevice.WANDeviceNumberOfEntries : 0 +InternetGatewayDevice.DeviceInfo. : 0 +InternetGatewayDevice.ManagementServer. : 0 +InternetGatewayDevice.Time. : 0 +InternetGatewayDevice.Layer3Forwarding. : 0 +InternetGatewayDevice.LANDevice. : 0 +InternetGatewayDevice.WANDevice. : 0 +InternetGatewayDevice.X_00507F_InternetAcc. : 0 +InternetGatewayDevice.X_00507F_LAN. : 0 +InternetGatewayDevice.X_00507F_NAT. : 0 +InternetGatewayDevice.X_00507F_VLAN. : 0 +InternetGatewayDevice.X_00507F_Firewall. : 0 +InternetGatewayDevice.X_00507F_Applications. : 0 +InternetGatewayDevice.X_00507F_System. : 0 +InternetGatewayDevice.X_00507F_Status. : 0 +InternetGatewayDevice.X_00507F_Diagnostics. : 0 +``` + + + + +## Services exposed + +Moses exposes three services: + + - http://localhost:9292/acs is the endpoint for the CPEs to connect + - http://localhost:9292/www is the embedded webserver to control your CPEs + - ws://localhost:9292/ws is the websocket endpoint used by the cli to issue commands. Read about the API specification if you want to build a custom frontend which interacts with mosesacs daemon. + + diff --git a/backend/services/acs/cmd/main.go b/backend/services/acs/cmd/main.go new file mode 100644 index 0000000..3ab03b9 --- /dev/null +++ b/backend/services/acs/cmd/main.go @@ -0,0 +1,19 @@ +package main + +import ( + "oktopUSP/backend/services/acs/internal/config" + "oktopUSP/backend/services/acs/internal/nats" + "oktopUSP/backend/services/acs/internal/server" + "oktopUSP/backend/services/acs/internal/server/handler" +) + +func main() { + + c := config.NewConfig() + + natsActions := nats.StartNatsClient(c.Nats) + + h := handler.NewHandler(natsActions.Publish, natsActions.Subscribe) + + server.Run(c.Acs, natsActions, h) +} diff --git a/backend/services/acs/go.mod b/backend/services/acs/go.mod new file mode 100644 index 0000000..5ba5c09 --- /dev/null +++ b/backend/services/acs/go.mod @@ -0,0 +1,19 @@ +module oktopUSP/backend/services/acs + +go 1.22.2 + +require ( + github.com/joho/godotenv v1.5.1 + github.com/nats-io/nats.go v1.34.1 + github.com/oleiade/lane v1.0.1 + golang.org/x/net v0.24.0 +) + +require ( + github.com/klauspost/compress v1.17.2 // indirect + github.com/nats-io/nkeys v0.4.7 // indirect + github.com/nats-io/nuid v1.0.1 // indirect + golang.org/x/crypto v0.22.0 // indirect + golang.org/x/sys v0.19.0 // indirect + golang.org/x/text v0.14.0 // indirect +) diff --git a/backend/services/acs/go.sum b/backend/services/acs/go.sum new file mode 100644 index 0000000..f09390a --- /dev/null +++ b/backend/services/acs/go.sum @@ -0,0 +1,20 @@ +github.com/joho/godotenv v1.5.1 h1:7eLL/+HRGLY0ldzfGMeQkb7vMd0as4CfYvUVzLqw0N0= +github.com/joho/godotenv v1.5.1/go.mod h1:f4LDr5Voq0i2e/R5DDNOoa2zzDfwtkZa6DnEwAbqwq4= +github.com/klauspost/compress v1.17.2 h1:RlWWUY/Dr4fL8qk9YG7DTZ7PDgME2V4csBXA8L/ixi4= +github.com/klauspost/compress v1.17.2/go.mod h1:ntbaceVETuRiXiv4DpjP66DpAtAGkEQskQzEyD//IeE= +github.com/nats-io/nats.go v1.34.1 h1:syWey5xaNHZgicYBemv0nohUPPmaLteiBEUT6Q5+F/4= +github.com/nats-io/nats.go v1.34.1/go.mod h1:Ubdu4Nh9exXdSz0RVWRFBbRfrbSxOYd26oF0wkWclB8= +github.com/nats-io/nkeys v0.4.7 h1:RwNJbbIdYCoClSDNY7QVKZlyb/wfT6ugvFCiKy6vDvI= +github.com/nats-io/nkeys v0.4.7/go.mod h1:kqXRgRDPlGy7nGaEDMuYzmiJCIAAWDK0IMBtDmGD0nc= +github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw= +github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c= +github.com/oleiade/lane v1.0.1 h1:hXofkn7GEOubzTwNpeL9MaNy8WxolCYb9cInAIeqShU= +github.com/oleiade/lane v1.0.1/go.mod h1:IyTkraa4maLfjq/GmHR+Dxb4kCMtEGeb+qmhlrQ5Mk4= +golang.org/x/crypto v0.22.0 h1:g1v0xeRhjcugydODzvb3mEM9SQ0HGp9s/nh3COQ/C30= +golang.org/x/crypto v0.22.0/go.mod h1:vr6Su+7cTlO45qkww3VDJlzDn0ctJvRgYbC2NvXHt+M= +golang.org/x/net v0.24.0 h1:1PcaxkF854Fu3+lvBIx5SYn9wRlBzzcnHZSiaFFAb0w= +golang.org/x/net v0.24.0/go.mod h1:2Q7sJY5mzlzWjKtYUEXSlBWCdyaioyXzRB2RtU8KVE8= +golang.org/x/sys v0.19.0 h1:q5f1RH2jigJ1MoAWp2KTp3gm5zAGFUTarQZ5U386+4o= +golang.org/x/sys v0.19.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= +golang.org/x/text v0.14.0 h1:ScX5w1eTa3QqT8oi6+ziP7dTV1S2+ALU0bI+0zXKWiQ= +golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU= diff --git a/backend/services/acs/internal/auth/auth.go b/backend/services/acs/internal/auth/auth.go new file mode 100644 index 0000000..476936f --- /dev/null +++ b/backend/services/acs/internal/auth/auth.go @@ -0,0 +1,113 @@ +package auth + +import ( + "crypto/md5" + "crypto/rand" + "encoding/base64" + "fmt" + "io" + "net/http" + "net/url" + "strings" +) + +type myjar struct { + jar map[string][]*http.Cookie +} + +func (p *myjar) SetCookies(u *url.URL, cookies []*http.Cookie) { + p.jar[u.Host] = cookies +} + +func (p *myjar) Cookies(u *url.URL) []*http.Cookie { + return p.jar[u.Host] +} + +func Auth(username string, password string, uri string) (bool, error) { + client := &http.Client{} + jar := &myjar{} + jar.jar = make(map[string][]*http.Cookie) + client.Jar = jar + var req *http.Request + var resp *http.Response + var err error + req, err = http.NewRequest("GET", uri, nil) + resp, err = client.Do(req) + if err != nil { + return false, err + } + if resp.StatusCode == 401 { + var authorization map[string]string = DigestAuthParams(resp) + realmHeader := authorization["realm"] + qopHeader := authorization["qop"] + nonceHeader := authorization["nonce"] + opaqueHeader := authorization["opaque"] + realm := realmHeader + // A1 + h := md5.New() + A1 := fmt.Sprintf("%s:%s:%s", username, realm, password) + io.WriteString(h, A1) + HA1 := fmt.Sprintf("%x", h.Sum(nil)) + + // A2 + h = md5.New() + A2 := fmt.Sprintf("GET:%s", "/auth") + io.WriteString(h, A2) + HA2 := fmt.Sprintf("%x", h.Sum(nil)) + + // response + cnonce := RandomKey() + response := H(strings.Join([]string{HA1, nonceHeader, "00000001", cnonce, qopHeader, HA2}, ":")) + + // now make header + AuthHeader := fmt.Sprintf(`Digest username="%s", realm="%s", nonce="%s", uri="%s", cnonce="%s", nc=00000001, qop=%s, response="%s", opaque="%s", algorithm=MD5`, + username, realmHeader, nonceHeader, "/auth", cnonce, qopHeader, response, opaqueHeader) + req.Header.Set("Authorization", AuthHeader) + resp, err = client.Do(req) + } else { + return false, fmt.Errorf("response status code should have been 401, it was %v", resp.StatusCode) + } + return resp.StatusCode == 200, err +} + +/* +Parse Authorization header from the http.Request. Returns a map of +auth parameters or nil if the header is not a valid parsable Digest +auth header. +*/ +func DigestAuthParams(r *http.Response) map[string]string { + s := strings.SplitN(r.Header.Get("Www-Authenticate"), " ", 2) + if len(s) != 2 || s[0] != "Digest" { + return nil + } + + result := map[string]string{} + for _, kv := range strings.Split(s[1], ",") { + parts := strings.SplitN(kv, "=", 2) + if len(parts) != 2 { + continue + } + result[strings.Trim(parts[0], "\" ")] = strings.Trim(parts[1], "\" ") + } + return result +} +func RandomKey() string { + k := make([]byte, 12) + for bytes := 0; bytes < len(k); { + n, err := rand.Read(k[bytes:]) + if err != nil { + panic("rand.Read() failed") + } + bytes += n + } + return base64.StdEncoding.EncodeToString(k) +} + +/* +H function for MD5 algorithm (returns a lower-case hex MD5 digest) +*/ +func H(data string) string { + digest := md5.New() + digest.Write([]byte(data)) + return fmt.Sprintf("%x", digest.Sum(nil)) +} diff --git a/backend/services/acs/internal/config/config.go b/backend/services/acs/internal/config/config.go new file mode 100644 index 0000000..9ad7d24 --- /dev/null +++ b/backend/services/acs/internal/config/config.go @@ -0,0 +1,111 @@ +package config + +import ( + "context" + "flag" + "log" + "os" + "strconv" + + "github.com/joho/godotenv" +) + +const LOCAL_ENV = ".env.local" + +type Nats struct { + Url string + Name string + VerifyCertificates bool + Ctx context.Context +} + +type Acs struct { + Port string + Tls bool + TlsPort bool + NoTls bool + Username string + Password string + Route string +} + +type Config struct { + Acs Acs + Nats Nats +} + +func NewConfig() *Config { + + loadEnvVariables() + log.SetFlags(log.LstdFlags | log.Lshortfile) + + natsUrl := flag.String("nats_url", lookupEnvOrString("NATS_URL", "nats://localhost:4222"), "url for nats server") + natsName := flag.String("nats_name", lookupEnvOrString("NATS_NAME", "adapter"), "name for nats client") + natsVerifyCertificates := flag.Bool("nats_verify_certificates", lookupEnvOrBool("NATS_VERIFY_CERTIFICATES", false), "verify validity of certificates from nats server") + acsPort := flag.String("acs_port", lookupEnvOrString("ACS_PORT", ":9292"), "port for acs server") + acsRoute := flag.String("acs_route", lookupEnvOrString("ACS_ROUTE", "/acs"), "route for acs server") + flHelp := flag.Bool("help", false, "Help") + + /* + App variables priority: + 1º - Flag through command line. + 2º - Env variables. + 3º - Default flag value. + */ + + flag.Parse() + + if *flHelp { + flag.Usage() + os.Exit(0) + } + + ctx := context.TODO() + + return &Config{ + Nats: Nats{ + Url: *natsUrl, + Name: *natsName, + VerifyCertificates: *natsVerifyCertificates, + Ctx: ctx, + }, + Acs: Acs{ + Port: *acsPort, + Route: *acsRoute, + }, + } +} + +func loadEnvVariables() { + err := godotenv.Load() + + if _, err := os.Stat(LOCAL_ENV); err == nil { + _ = godotenv.Overload(LOCAL_ENV) + log.Printf("Loaded variables from '%s'", LOCAL_ENV) + return + } + + if err != nil { + log.Println("Error to load environment variables:", err) + } else { + log.Println("Loaded variables from '.env'") + } +} + +func lookupEnvOrString(key string, defaultVal string) string { + if val, _ := os.LookupEnv(key); val != "" { + return val + } + return defaultVal +} + +func lookupEnvOrBool(key string, defaultVal bool) bool { + if val, _ := os.LookupEnv(key); val != "" { + v, err := strconv.ParseBool(val) + if err != nil { + log.Fatalf("LookupEnvOrBool[%s]: %v", key, err) + } + return v + } + return defaultVal +} diff --git a/backend/services/acs/internal/cwmp/cwmp.go b/backend/services/acs/internal/cwmp/cwmp.go new file mode 100644 index 0000000..4d69201 --- /dev/null +++ b/backend/services/acs/internal/cwmp/cwmp.go @@ -0,0 +1,541 @@ +package cwmp + +import ( + "crypto/rand" + "encoding/xml" + "fmt" + "strconv" + "strings" + "time" +) + +type SoapEnvelope struct { + XMLName xml.Name + Header SoapHeader + Body SoapBody +} + +type SoapHeader struct { + Id string `xml:"ID"` +} +type SoapBody struct { + CWMPMessage CWMPMessage `xml:",any"` +} + +type CWMPMessage struct { + XMLName xml.Name +} + +type EventStruct struct { + EventCode string + CommandKey string +} + +type ParameterValueStruct struct { + Name string + Value string +} + +type ParameterInfoStruct struct { + Name string + Writable string +} + +type SetParameterValues_ struct { + ParameterList []ParameterValueStruct `xml:"Body>SetParameterValues>ParameterList>ParameterValueStruct"` + ParameterKey string `xml:"Body>SetParameterValues>ParameterKey>string"` +} + +type GetParameterValues_ struct { + ParameterNames []string `xml:"Body>GetParameterValues>ParameterNames>string"` +} + +type GetParameterNames_ struct { + ParameterPath []string `xml:"Body>GetParameterNames>ParameterPath"` + NextLevel string `xml:"Body>GetParameterNames>NextLevel"` +} + +type GetParameterValuesResponse struct { + ParameterList []ParameterValueStruct `xml:"Body>GetParameterValuesResponse>ParameterList>ParameterValueStruct"` +} + +type GetParameterNamesResponse struct { + ParameterList []ParameterInfoStruct `xml:"Body>GetParameterNamesResponse>ParameterList>ParameterInfoStruct"` +} + +type CWMPInform struct { + DeviceId DeviceID `xml:"Body>Inform>DeviceId"` + Events []EventStruct `xml:"Body>Inform>Event>EventStruct"` + ParameterList []ParameterValueStruct `xml:"Body>Inform>ParameterList>ParameterValueStruct"` +} + +func (s *SoapEnvelope) KindOf() string { + return s.Body.CWMPMessage.XMLName.Local +} + +func (i *CWMPInform) GetEvents() string { + res := "" + for idx := range i.Events { + res += i.Events[idx].EventCode + } + + return res +} + +func (i *CWMPInform) GetConnectionRequest() string { + for idx := range i.ParameterList { + // valid condition for both tr98 and tr181 + if strings.HasSuffix(i.ParameterList[idx].Name, "Device.ManagementServer.ConnectionRequestURL") { + return i.ParameterList[idx].Value + } + } + + return "" +} + +func (i *CWMPInform) GetSoftwareVersion() string { + for idx := range i.ParameterList { + if strings.HasSuffix(i.ParameterList[idx].Name, "Device.DeviceInfo.SoftwareVersion") { + return i.ParameterList[idx].Value + } + } + + return "" +} + +func (i *CWMPInform) GetHardwareVersion() string { + for idx := range i.ParameterList { + if strings.HasSuffix(i.ParameterList[idx].Name, "Device.DeviceInfo.HardwareVersion") { + return i.ParameterList[idx].Value + } + } + + return "" +} + +func (i *CWMPInform) GetDataModelType() string { + if strings.HasPrefix(i.ParameterList[0].Name, "InternetGatewayDevice") { + return "TR098" + } else if strings.HasPrefix(i.ParameterList[0].Name, "Device") { + return "TR181" + } + + return "" +} + +type DeviceID struct { + Manufacturer string + OUI string + SerialNumber string +} + +func InformResponse(mustUnderstand string) string { + mustUnderstandHeader := "" + if mustUnderstand != "" { + mustUnderstandHeader = `` + mustUnderstand + `` + } + + return ` + + ` + mustUnderstandHeader + ` + + + 1 + + +` +} + +func GetParameterValues(leaf string) string { + return ` + + + + + + ` + leaf + ` + + + +` +} + +func GetParameterMultiValues(leaves []string) string { + msg := ` + + + + + ` + + for idx := range leaves { + msg += `` + leaves[idx] + `` + + } + msg += ` + + +` + return msg +} + +func SetParameterValues(leaf string, value string) string { + return ` + + + + + + + ` + leaf + ` + ` + value + ` + + + LC1309` + randToken() + ` + + +` +} + +func randToken() string { + b := make([]byte, 8) + rand.Read(b) + return fmt.Sprintf("%x", b) +} + +func SetParameterMultiValues(data map[string]string) string { + msg := ` + + + + + ` + + for key, value := range data { + msg += ` + ` + key + ` + ` + value + ` + ` + } + + msg += ` + LC1309` + randToken() + ` + + +` + + return msg +} + +func GetParameterNames(leaf string, nextlevel int) string { + return ` + + + + + ` + leaf + ` + ` + strconv.Itoa(nextlevel) + ` + + +` +} + +func FactoryReset() string { + return ` + + + + + +` +} + +func Download(filetype, url, username, password, filesize string) string { + // 3 Vendor Configuration File + // 1 Firmware Upgrade Image + + return ` + + + + + MSDWK + ` + filetype + ` + ` + url + ` + ` + username + ` + ` + password + ` + ` + filesize + ` + + 0 + + + + +` +} + +func CancelTransfer() string { + return ` + + + + + + + +` +} + +type TimeWindowStruct struct { + WindowStart string + WindowEnd string + WindowMode string + UserMessage string + MaxRetries string +} + +func (window *TimeWindowStruct) String() string { + return ` +` + window.WindowStart + ` +` + window.WindowEnd + ` +` + window.WindowMode + ` +` + window.UserMessage + ` +` + window.MaxRetries + ` +` +} + +func ScheduleDownload(filetype, url, username, password, filesize string, windowslist []fmt.Stringer) string { + ret := ` + + + + + MSDWK + ` + filetype + ` + ` + url + ` + ` + username + ` + ` + password + ` + ` + filesize + ` + + ` + + for _, op := range windowslist { + ret += op.String() + } + + ret += ` + + +` + + return ret +} + +type InstallOpStruct struct { + Url string + Uuid string + Username string + Password string + ExecutionEnvironment string +} + +func (op *InstallOpStruct) String() string { + return ` + ` + op.Url + ` + ` + op.Uuid + ` + ` + op.Username + ` + ` + op.Password + ` + ` + op.ExecutionEnvironment + ` +` +} + +type UpdateOpStruct struct { + Uuid string + Version string + Url string + Username string + Password string +} + +func (op *UpdateOpStruct) String() string { + return ` +` + op.Uuid + ` +` + op.Version + ` +` + op.Url + ` +` + op.Username + ` +` + op.Password + ` +` +} + +type UninstallOpStruct struct { + Uuid string + Version string + ExecutionEnvironment string +} + +func (op *UninstallOpStruct) String() string { + return ` +` + op.Uuid + ` +` + op.Version + ` +` + op.ExecutionEnvironment + ` +` +} + +func ChangeDuState(ops []fmt.Stringer) string { + ret := ` + + + + +` + + for _, op := range ops { + ret += op.String() + } + + ret += ` + + + +` + + return ret +} + +// CPE side + +func Inform(serial string) string { + return `5058 + ADB Broadband +0013C8 +VV5522 +` + serial + ` + + +6 CONNECTION REQUEST + + + +1 +` + time.Now().Format(time.RFC3339) + ` +0 + +InternetGatewayDevice.ManagementServer.ConnectionRequestURL +http://104.199.175.27:7547/ConnectionRequest-` + serial + ` + +InternetGatewayDevice.ManagementServer.ParameterKey + + +InternetGatewayDevice.DeviceSummary +InternetGatewayDevice:1.2[](Baseline:1,EthernetLAN:1,WiFiLAN:1,ADSLWAN:1,EthernetWAN:1,QoS:1,QoSDynamicFlow:1,Bridging:1,Time:1,IPPing:1,TraceRoute:1,DeviceAssociation:1,UDPConnReq:1),VoiceService:1.0[1](TAEndpoint:1,SIPEndpoint:1) + +InternetGatewayDevice.DeviceInfo.HardwareVersion +` + serial + ` + +InternetGatewayDevice.DeviceInfo.ProvisioningCode +ABCD + +InternetGatewayDevice.DeviceInfo.SoftwareVersion +4.0.8.17785 + +InternetGatewayDevice.DeviceInfo.SpecVersion +1.0 + +InternetGatewayDevice.WANDevice.1.WANConnectionDevice.1.WANIPConnection.1.ExternalIPAddress +12.0.0.10 + + + +` +} + +/* +func BuildGetParameterValuesResponse(serial string, leaves GetParameterValues_) string { + ret := ` + 3 + ` + + db, _ := sqlite3.Open("/tmp/cpe.db") + + n_leaves := 0 + var temp string + for _, leaf := range leaves.ParameterNames { + sql := "select key, value, tipo from params where key like '" + leaf + "%'" + for s, err := db.Query(sql); err == nil; err = s.Next() { + n_leaves++ + var key string + var value string + var tipo string + s.Scan(&key, &value, &tipo) + temp += ` + ` + key + ` + ` + value + ` + ` + } + } + + ret += `` + ret += temp + ret += `` + + return ret +} + +func BuildGetParameterNamesResponse(serial string, leaves GetParameterNames_) string { + ret := ` + 69 + ` + db, _ := sqlite3.Open("/tmp/cpe.db") + + obj := make(map[string]bool) + var temp string + for _, leaf := range leaves.ParameterPath { + fmt.Println(leaf) + sql := "select key, value, tipo from params where key like '" + leaf + "%'" + for s, err := db.Query(sql); err == nil; err = s.Next() { + var key string + var value string + var tipo string + s.Scan(&key, &value, &tipo) + var sp = strings.Split(strings.Split(key, leaf)[1], ".") + nextlevel, _ := strconv.Atoi(leaves.NextLevel) + if nextlevel == 0 { + root := leaf + obj[root] = true + for idx := range sp { + if idx == len(sp)-1 { + root = root + sp[idx] + } else { + root = root + sp[idx] + "." + } + obj[root] = true + } + } else { + if !obj[sp[0]] { + if len(sp) > 1 { + obj[leaf+sp[0]+"."] = true + } else { + obj[leaf+sp[0]] = true + } + + } + } + + } + } + + for o := range obj { + temp += ` + ` + o + ` + true + ` + } + + fmt.Println(len(obj)) + ret += `` + ret += temp + ret += `` + + return ret +} +*/ diff --git a/backend/services/acs/internal/nats/nats.go b/backend/services/acs/internal/nats/nats.go new file mode 100644 index 0000000..0b1e803 --- /dev/null +++ b/backend/services/acs/internal/nats/nats.go @@ -0,0 +1,149 @@ +package nats + +import ( + "context" + "errors" + "log" + "oktopUSP/backend/services/acs/internal/config" + "time" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" +) + +const ( + CWMP_STREAM_NAME = "cwmp" +) + +type NatsActions struct { + Publish func(string, []byte) error + Subscribe func(string, func(*nats.Msg)) error +} + +func StartNatsClient(c config.Nats) NatsActions { + + var ( + nc *nats.Conn + err error + ) + + opts := defineOptions(c) + + log.Printf("Connecting to NATS server %s", c.Url) + + for { + nc, err = nats.Connect(c.Url, opts...) + if err != nil { + time.Sleep(5 * time.Second) + continue + } + break + } + log.Printf("Successfully connected to NATS server %s", c.Url) + + js, err := jetstream.New(nc) + if err != nil { + log.Fatalf("Failed to create JetStream client: %v", err) + } + + streams := defineStreams() + err = createStreams(c.Ctx, js, streams) + if err != nil { + log.Fatalf("Failed to create Consumer: %v", err) + } + + consumers := defineConsumers() + err = createConsumers(c.Ctx, js, consumers) + if err != nil { + log.Fatalf("Failed to create Consumer: %v", err) + } + + return NatsActions{ + Publish: publisher(js), + Subscribe: subscriber(nc), + } +} + +func subscriber(nc *nats.Conn) func(string, func(*nats.Msg)) error { + return func(subject string, handler func(*nats.Msg)) error { + _, err := nc.Subscribe(subject, handler) + if err != nil { + log.Printf("error to subscribe to subject %s error: %q", subject, err) + } + return err + } +} + +func publisher(js jetstream.JetStream) func(string, []byte) error { + return func(subject string, payload []byte) error { + _, err := js.PublishAsync(subject, payload) + if err != nil { + log.Printf("error to send jetstream message: %q", err) + } + return err + } +} + +func createStreams(ctx context.Context, js jetstream.JetStream, streams []string) error { + for _, stream := range streams { + _, err := js.CreateOrUpdateStream(ctx, jetstream.StreamConfig{ + Name: stream, + Description: "Stream for " + stream + " messages", + Subjects: []string{stream + ".>"}, + Retention: jetstream.InterestPolicy, + }) + if err != nil { + return errors.New(err.Error() + " | consumer:" + stream) + } + } + return nil +} + +func createConsumers(ctx context.Context, js jetstream.JetStream, consumers []string) error { + for _, consumer := range consumers { + _, err := js.CreateOrUpdateConsumer(ctx, consumer, jetstream.ConsumerConfig{ + Name: consumer, + Description: "Consumer for " + consumer + " messages", + AckPolicy: jetstream.AckExplicitPolicy, + Durable: consumer, + }) + if err != nil { + return err + } + } + return nil +} + +func defineStreams() []string { + return []string{ + CWMP_STREAM_NAME, + } +} + +func defineConsumers() []string { + return []string{ + CWMP_STREAM_NAME, + } +} + +func defineOptions(c config.Nats) []nats.Option { + var opts []nats.Option + + opts = append(opts, nats.Name(c.Name)) + opts = append(opts, nats.MaxReconnects(-1)) + opts = append(opts, nats.ReconnectWait(5*time.Second)) + opts = append(opts, nats.DisconnectErrHandler(func(nc *nats.Conn, err error) { + log.Printf("Got disconnected! Reason: %q\n", err) + })) + opts = append(opts, nats.ReconnectHandler(func(nc *nats.Conn) { + log.Printf("Got reconnected to %v!\n", nc.ConnectedUrl()) + })) + opts = append(opts, nats.ClosedHandler(func(nc *nats.Conn) { + log.Printf("Connection closed. Reason: %q\n", nc.LastError()) + })) + if c.VerifyCertificates { + opts = append(opts, nats.RootCAs()) + } + + return opts +} diff --git a/backend/services/acs/internal/server/handler/handler.go b/backend/services/acs/internal/server/handler/handler.go new file mode 100644 index 0000000..e86e28a --- /dev/null +++ b/backend/services/acs/internal/server/handler/handler.go @@ -0,0 +1,215 @@ +package handler + +import ( + "encoding/json" + "encoding/xml" + "fmt" + "io/ioutil" + "log" + "net/http" + "oktopUSP/backend/services/acs/internal/cwmp" + "time" + + "github.com/nats-io/nats.go" + "github.com/oleiade/lane" + "golang.org/x/net/websocket" +) + +const Version = "1.0.0" + +type Request struct { + Id string + Websocket *websocket.Conn + CwmpMessage string + Callback func(msg *WsSendMessage) error +} + +type CPE struct { + SerialNumber string + Manufacturer string + OUI string + ConnectionRequestURL string + XmppId string + XmppUsername string + XmppPassword string + SoftwareVersion string + ExternalIPAddress string + State string + Queue *lane.Queue + Waiting *Request + HardwareVersion string + LastConnection time.Time + DataModel string + KeepConnectionOpen bool +} + +type Message struct { + SerialNumber string + Message string +} + +type WsMessage struct { + Cmd string +} + +type WsSendMessage struct { + MsgType string + Data json.RawMessage +} + +type MsgCPEs struct { + CPES map[string]CPE +} + +type Handler struct { + pub func(string, []byte) error + sub func(string, func(*nats.Msg)) error + cpes map[string]CPE +} + +func NewHandler(pub func(string, []byte) error, sub func(string, func(*nats.Msg)) error) *Handler { + return &Handler{ + pub: pub, + sub: sub, + cpes: make(map[string]CPE), + } +} + +func (h *Handler) CwmpHandler(w http.ResponseWriter, r *http.Request) { + + log.Printf("--> Connection from %s", r.RemoteAddr) + + defer r.Body.Close() + defer log.Printf("<-- Connection from %s closed", r.RemoteAddr) + + tmp, _ := ioutil.ReadAll(r.Body) + body := string(tmp) + len := len(body) + + log.Printf("body:\n %v", body) + + var envelope cwmp.SoapEnvelope + xml.Unmarshal(tmp, &envelope) + + messageType := envelope.Body.CWMPMessage.XMLName.Local + + var cpe *CPE + + w.Header().Set("Server", "Oktopus "+Version) + + if messageType != "Inform" { + if cookie, err := r.Cookie("oktopus"); err == nil { + if _, exists := h.cpes[cookie.Value]; !exists { + log.Printf("CPE with serial number %s not found", cookie.Value) + } + } else { + fmt.Println("cookie 'oktopus' missing") + w.WriteHeader(401) + return + } + } + + if messageType == "Inform" { + var Inform cwmp.CWMPInform + xml.Unmarshal(tmp, &Inform) + + var addr string + if r.Header.Get("X-Real-Ip") != "" { + addr = r.Header.Get("X-Real-Ip") + } else { + addr = r.RemoteAddr + } + + sn := Inform.DeviceId.SerialNumber + + if _, exists := h.cpes[sn]; !exists { + fmt.Println("New device: " + sn) + h.cpes[sn] = CPE{ + SerialNumber: sn, + LastConnection: time.Now().UTC(), + SoftwareVersion: Inform.GetSoftwareVersion(), + HardwareVersion: Inform.GetHardwareVersion(), + ExternalIPAddress: addr, + ConnectionRequestURL: Inform.GetConnectionRequest(), + OUI: Inform.DeviceId.OUI, + Queue: lane.NewQueue(), + DataModel: Inform.GetDataModelType(), + KeepConnectionOpen: false} + } + obj := h.cpes[sn] + cpe := &obj + cpe.LastConnection = time.Now().UTC() + + log.Printf("Received an Inform from %s (%d bytes) with SerialNumber %s and EventCodes %s", addr, len, sn, Inform.GetEvents()) + + expiration := time.Now().AddDate(0, 0, 1) + + cookie := http.Cookie{Name: "oktopus", Value: sn, Expires: expiration} + http.SetCookie(w, &cookie) + } else if messageType == "TransferComplete" { + + } else if messageType == "GetRPC" { + + } else { + if len == 0 { + log.Printf("Got Empty Post") + } + + if cpe.Waiting != nil { + var e cwmp.SoapEnvelope + xml.Unmarshal([]byte(body), &e) + + if e.KindOf() == "GetParameterNamesResponse" { + var envelope cwmp.GetParameterNamesResponse + xml.Unmarshal([]byte(body), &envelope) + + msg := new(WsSendMessage) + msg.MsgType = "GetParameterNamesResponse" + msg.Data, _ = json.Marshal(envelope) + + cpe.Waiting.Callback(msg) + // if err := websocket.JSON.Send(cpe.Waiting.Websocket, msg); err != nil { + // fmt.Println("error while sending back answer:", err) + // } + + } else if e.KindOf() == "GetParameterValuesResponse" { + var envelope cwmp.GetParameterValuesResponse + xml.Unmarshal([]byte(body), &envelope) + + msg := new(WsSendMessage) + msg.MsgType = "GetParameterValuesResponse" + msg.Data, _ = json.Marshal(envelope) + + cpe.Waiting.Callback(msg) + // if err := websocket.JSON.Send(cpe.Waiting.Websocket, msg); err != nil { + // fmt.Println("error while sending back answer:", err) + // } + + } else { + msg := new(WsMessage) + msg.Cmd = body + + if err := websocket.JSON.Send(cpe.Waiting.Websocket, msg); err != nil { + fmt.Println("error while sending back answer:", err) + } + + } + + cpe.Waiting = nil + } + + // Got Empty Post or a Response. Now check for any event to send, otherwise 204 + if cpe.Queue.Size() > 0 { + req := cpe.Queue.Dequeue().(Request) + // fmt.Println("sending "+req.CwmpMessage) + fmt.Fprintf(w, req.CwmpMessage) + cpe.Waiting = &req + } else { + if cpe.KeepConnectionOpen { + fmt.Println("I'm keeping connection open") + } else { + w.WriteHeader(204) + } + } + } +} diff --git a/backend/services/acs/internal/server/server.go b/backend/services/acs/internal/server/server.go new file mode 100644 index 0000000..4a2bc6b --- /dev/null +++ b/backend/services/acs/internal/server/server.go @@ -0,0 +1,23 @@ +package server + +import ( + "log" + "net/http" + "oktopUSP/backend/services/acs/internal/config" + "oktopUSP/backend/services/acs/internal/nats" + "oktopUSP/backend/services/acs/internal/server/handler" + "os" +) + +func Run(c config.Acs, natsActions nats.NatsActions, h *handler.Handler) { + + http.HandleFunc(c.Route, h.CwmpHandler) + + log.Printf("ACS running at %s%s", c.Port, c.Route) + + err := http.ListenAndServe(c.Port, nil) + if err != nil { + log.Fatal(err) + os.Exit(1) + } +} diff --git a/backend/services/mtp/adapter/.env.local b/backend/services/mtp/adapter/.env.local deleted file mode 100644 index c9b245e..0000000 --- a/backend/services/mtp/adapter/.env.local +++ /dev/null @@ -1,2 +0,0 @@ -MONGO_URI="mongodb://localhost:27017" -CONTROLLER_PASSWORD="test123" diff --git a/backend/services/mtp/adapter/.gitignore b/backend/services/mtp/adapter/.gitignore new file mode 100644 index 0000000..e45d6e3 --- /dev/null +++ b/backend/services/mtp/adapter/.gitignore @@ -0,0 +1 @@ +*.local \ No newline at end of file diff --git a/deploy/compose/.env.mqtt b/deploy/compose/.env.mqtt index 2939a97..e0c2d35 100644 --- a/deploy/compose/.env.mqtt +++ b/deploy/compose/.env.mqtt @@ -1,2 +1,3 @@ REDIS_ENABLE=false -REDIS_ADDR=redis_usp:6379 \ No newline at end of file +REDIS_ADDR=redis_usp:6379 +NATS_URL=nats://msg_broker:4222 \ No newline at end of file diff --git a/deploy/compose/.env.ws b/deploy/compose/.env.ws new file mode 100644 index 0000000..9af7128 --- /dev/null +++ b/deploy/compose/.env.ws @@ -0,0 +1 @@ +NATS_URL=nats://msg_broker:4222 \ No newline at end of file diff --git a/deploy/compose/docker-compose.yaml b/deploy/compose/docker-compose.yaml index b2433ac..acf57fb 100644 --- a/deploy/compose/docker-compose.yaml +++ b/deploy/compose/docker-compose.yaml @@ -66,6 +66,8 @@ services: container_name: websockets ports: - 8080:8080 + env_file: + - .env.ws networks: usp_network: ipv4_address: 172.16.235.7