Nats Native Behaviors + Peer is Stateless

This commit is contained in:
mr
2026-02-18 14:25:56 +01:00
parent fa5c3a3c60
commit 0b41e2505e
7 changed files with 280 additions and 42 deletions

190
UNUSED_AND_ISSUES.md Normal file
View File

@@ -0,0 +1,190 @@
# Rapport d'audit Éléments inutilisés et problèmes identifiés
> Généré le 2026-02-18 branche `feature/event`
---
## 1. Bugs critiques corrigés dans cette session
| Fichier | Ligne | Description | Statut |
|---------|-------|-------------|--------|
| `entrypoint.go` | 652, 664, 676, 688 | `fmt.Errorf(res.Err)` format string non-constant (erreur de build) | Corrigé |
| `models/utils/abstracts.go` | 136 | `VerifyAuth` déréférençait `request.Admin` avant de vérifier `request != nil` | Corrigé |
| `models/utils/abstracts.go` | 68-78 | `DeepCopy()` faisait `Unmarshal` dans un pointeur nil retournait toujours `nil` | Corrigé |
| `models/resources/resource.go` | 176 | `instances = append(instances)` argument manquant, l'instance n'était jamais ajoutée | Corrigé |
| `models/resources/priced_resource.go` | 63-69 | Code mort après `return true` dans `IsBooked()` | Corrigé |
| `tools/remote_caller.go` | 118 | `CallDelete` vérifiait `req.Body == nil` (toujours vrai pour DELETE), court-circuitant la lecture de la réponse | Corrigé |
---
## 2. Debug prints à supprimer (fmt.Println en production)
Ces appels `fmt.Println` polluent stdout et peuvent exposer des informations sensibles.
| Fichier | Lignes | Contenu |
|---------|--------|---------|
| `models/bill/bill.go` | ~197 | `fmt.Println(err)` |
| `models/collaborative_area/collaborative_area_mongo_accessor.go` | ~95, 109, 118, 123 | Debug sur `res`, `sharedWorkspace.AllowedPeersGroup`, `canFound`, `peerskey` |
| `models/peer/peer_cache.go` | ~44, 55 | URL et `"Launching peer execution on..."` |
| `models/resources/storage.go` | ~196 | `fmt.Println("GetPriceHT", ...)` |
| `models/workflow/workflow.go` | ~158, 164, 170, 176 | 4× `fmt.Println(err)` |
| `tools/nats_caller.go` | ~110, 117, 122, 126 | 4× `fmt.Println()` divers |
| `tools/remote_caller.go` | 227 | `fmt.Println("Error reading the body...")` (devrait utiliser le logger) |
| `dbs/dbs.go` | 47 | `fmt.Println("Recovered. Error:\n", r, debug.Stack())` |
> **Note :** `priced_resource.go` et `data.go` corrigés dans cette session.
---
## 3. Code commenté significatif
### 3.1 Validation de pricing désactivée (workflow)
**Fichier :** `models/workflow/workflow.go` ~lignes 631-634
```go
// Should be commented once the Pricing selection feature has been implemented
// if priced.SelectPricing() == nil {
// return resources, priceds, errors.New("no pricings are selected... can't proceed")
// }
```
Une vérification de sécurité critique est désactivée. Sans elle, des ressources sans pricing peuvent être traitées silencieusement.
### 3.2 PAY_PER_USE stratégie supprimée mais traces restantes
**Fichier :** `models/common/pricing/pricing_strategy.go` lignes 47, 61-63
```go
// PAY_PER_USE // per request. ( unpredictible )
/*case PAY_PER_USE:
return bs, true*/
```
La constante `PAY_PER_USE` a été supprimée mais les commentaires laissés créent de la confusion.
### 3.3 Vérification d'autorisation peer désactivée
**Fichier :** `models/resources/resource.go` lignes 98-104
```go
/*if ok, _ := utils.IsMySelf(request.PeerID, ...); ok {*/
profile = pricing.GetDefaultPricingProfile()
/*} else {
return nil, errors.New("no pricing profile found")
}*/
```
Le profil par défaut est retourné sans vérifier si le pair est bien `myself`. Sécurité à revoir.
---
## 4. Logique erronée non corrigée (à traiter)
### 4.1 IsTimeStrategy logique inversée
**Fichier :** `models/common/pricing/pricing_strategy.go` ligne 88
```go
func IsTimeStrategy(i int) bool {
return len(TimePricingStrategyList()) < i // BUG: devrait être ">"
}
```
La condition est inversée. Retourne `true` pour des valeurs hors de la liste. Fonction actuellement non utilisée (voir §5).
### 4.2 IsBillingStrategyAllowed case SUBSCRIPTION sans retour
**Fichier :** `models/common/pricing/pricing_strategy.go` lignes 54-65
```go
case SUBSCRIPTION:
/*case PAY_PER_USE:
return bs, true*/
// Aucun return ici → tombe dans le default
```
Le cas `SUBSCRIPTION` ne retourne rien explicitement, ce qui est trompeur.
---
## 5. Éléments inutilisés
### 5.1 Fonction jamais appelée
| Symbole | Fichier | Ligne |
|---------|---------|-------|
| `IsTimeStrategy(i int) bool` | `models/common/pricing/pricing_strategy.go` | 88 |
De plus, cette fonction a une logique erronée (voir §4.1).
### 5.2 Variable singleton inutilisée
| Symbole | Fichier | Ligne |
|---------|---------|-------|
| `HTTPCallerInstance` | `tools/remote_caller.go` | 57 |
Déclarée comme singleton mais jamais utilisée de nouvelles instances sont créées via `NewHTTPCaller()`.
---
## 6. Tests supprimés (couverture perdue)
Les fichiers suivants ont été supprimés sur la branche `feature/event` et la couverture correspondante n'est plus assurée :
| Fichier supprimé | Modèles non couverts |
|------------------|----------------------|
| `models/peer/tests/peer_cache_test.go` | `PeerCache` logique d'exécution distribuée |
| `models/peer/tests/peer_test.go` | `Peer` modèle et accesseur |
| `models/utils/tests/abstracts_test.go` | `AbstractObject` méthodes de base |
| `models/utils/tests/common_test.go` | `GenericStoreOne`, `GenericDeleteOne`, etc. |
| `models/workflow_execution/tests/workflow_test.go` | `WorkflowExecution` modèle et accesseur |
> `models/order/tests/order_test.go` existe mais ne contient **aucune fonction de test**.
---
## 7. Fautes d'orthographe dans les identifiants publics
Ces typos sont dans des noms exportés (API publique) les corriger est un **breaking change**.
### 7.1 `Instanciated` `Instantiated`
Apparaît 50+ fois dans les types exportés centraux :
- `AbstractInstanciatedResource[T]` (resource.go, compute.go, data.go, storage.go, processing.go, workflow.go)
- `AbstractInstanciatedResource.Instances`
- Tests : `resources.AbstractInstanciatedResource[*MockInstance]{...}`
### 7.2 `ressource` `resource` (dans les messages d'erreur)
**Fichier :** `entrypoint.go` messages dans `LoadOneStorage`, `LoadOneComputing`, etc.
```go
"Error while loading storage ressource " + storageId // "ressource" est du français
```
### 7.3 `GARANTED` `GUARANTEED`
**Fichiers :** `models/common/pricing/pricing_profile.go`, `models/resources/storage.go`
```go
GARANTED_ON_DELAY // pricing_profile.go:72
GARANTED // pricing_profile.go:73
GARANTED_ON_DELAY_STORAGE // storage.go:106
GARANTED_STORAGE // storage.go:107
```
### 7.4 `CREATE_EXECTUTION` `CREATE_EXECUTION`
**Fichier :** `tools/nats_caller.go` ligne 34
```go
CREATE_EXECTUTION // faute de frappe dans la constante enum
```
### 7.5 `PROPALGATION` `PROPAGATION`
**Fichier :** `tools/nats_caller.go` lignes 29, 45, 56
```go
"propalgation event" // et PROPALGATION_EVENT
```
---
## 8. Incohérences de nommage mineures
| Fichier | Problème |
|---------|----------|
| `models/resources/interfaces.go:19` | Paramètre `instance_id` en snake_case dans une signature Go (devrait être `instanceID`) |
| `entrypoint.go:505` | Message de panique dans `CopyOne` dit `"Panic recovered in UpdateOne"` |
| `tools/remote_caller.go:110` | Commentaire `// CallPut calls the DELETE method` (copie-colle incorrect) |
---
## 9. Résumé
| Catégorie | Nombre | Priorité |
|-----------|--------|----------|
| Bugs critiques corrigés | 6 | Fait |
| Debug `fmt.Println` restants | 15+ | 🔴 Haute |
| Code commenté important | 3 | 🟠 Moyenne |
| Logique erronée (non corrigée) | 2 | 🟠 Moyenne |
| Éléments inutilisés | 2 | 🟡 Faible |
| Tests supprimés (couverture perdue) | 5 fichiers | 🟠 Moyenne |
| Typos dans API publique | 5 types | 🟡 Faible (breaking change) |
| Incohérences mineures | 3 | 🟢 Très faible |

View File

@@ -7,6 +7,7 @@ import (
"fmt" "fmt"
"net/http" "net/http"
"os" "os"
"slices"
"strings" "strings"
"runtime/debug" "runtime/debug"
@@ -34,6 +35,7 @@ import (
"github.com/beego/beego/v2/server/web/filter/cors" "github.com/beego/beego/v2/server/web/filter/cors"
"github.com/google/uuid" "github.com/google/uuid"
"github.com/goraz/onion" "github.com/goraz/onion"
"github.com/libp2p/go-libp2p/core/crypto"
"github.com/rs/zerolog" "github.com/rs/zerolog"
) )
@@ -641,7 +643,7 @@ func GetConcatenatedName(peerId string, namespace string) string {
return peerId + "-" + n return peerId + "-" + n
} }
// ------------- Loading resources ---------- // ------------- Loading resources ----------GetAccessor
func LoadOneStorage(storageId string, user string, peerID string, groups []string) (*resources.StorageResource, error) { func LoadOneStorage(storageId string, user string, peerID string, groups []string) (*resources.StorageResource, error) {
@@ -690,3 +692,76 @@ func LoadOneData(dataId string, user string, peerID string, groups []string) (*r
return res.ToDataResource(), nil return res.ToDataResource(), nil
} }
// verify signature...
func InitNATSDecentralizedEmitter(authorizedDT ...tools.DataType) {
tools.NewNATSCaller().ListenNats(map[tools.NATSMethod]func(tools.NATSResponse){
tools.CREATE_RESOURCE: func(resp tools.NATSResponse) {
if resp.FromApp == config.GetAppName() || !slices.Contains(authorizedDT, resp.Datatype) {
return
}
p := map[string]interface{}{}
if err := json.Unmarshal(resp.Payload, &p); err == nil {
if err := verify(resp.Payload); err != nil {
return // don't trust anyone... only friends and foes are privilege
}
access := NewRequestAdmin(LibDataEnum(resp.Datatype), nil)
if data := access.Search(nil, fmt.Sprintf("%v", p[resp.SearchAttr]), false); len(data.Data) > 0 {
delete(p, "id")
access.UpdateOne(p, data.Data[0].GetID())
} else {
access.StoreOne(p)
}
}
},
tools.REMOVE_RESOURCE: func(resp tools.NATSResponse) {
if resp.FromApp == config.GetAppName() || !slices.Contains(authorizedDT, resp.Datatype) {
return
}
if err := verify(resp.Payload); err != nil {
return // don't trust anyone... only friends and foes are privilege
}
p := map[string]interface{}{}
access := NewRequestAdmin(LibDataEnum(resp.Datatype), nil)
err := json.Unmarshal(resp.Payload, &p)
if err == nil {
if data := access.Search(nil, fmt.Sprintf("%v", p[resp.SearchAttr]), false); len(data.Data) > 0 {
access.DeleteOne(fmt.Sprintf("%v", p[resp.SearchAttr]))
}
}
},
})
}
func verify(payload []byte) error {
var obj utils.AbstractObject
if err := json.Unmarshal(payload, &obj); err == nil {
obj.Unsign()
origin := NewRequestAdmin(LibDataEnum(PEER), nil).LoadOne(obj.GetCreatorID())
if origin.Data == nil || origin.Data.(*peer.Peer).Relation != peer.PARTNER {
return errors.New("don't know personnaly this guy") // don't trust anyone... only friends and foes are privilege
}
data, err := base64.StdEncoding.DecodeString(origin.Data.(*peer.Peer).PublicKey)
if err != nil {
return err
}
pk, err := crypto.UnmarshalPublicKey(data)
if err != nil {
return err
}
b, err := json.Marshal(obj)
if err != nil {
return err
}
if ok, err := pk.Verify(b, obj.GetSignature()); err != nil {
return err
} else if !ok {
return errors.New("signature is not corresponding to public key")
} else {
return nil
}
} else {
return err
}
}

View File

@@ -8,22 +8,6 @@ import (
"cloud.o-forge.io/core/oc-lib/tools" "cloud.o-forge.io/core/oc-lib/tools"
) )
// now write a go enum for the state partner with self, blacklist, partner
type PeerState int
const (
OFFLINE PeerState = iota
ONLINE
)
func (m PeerState) String() string {
return [...]string{"NONE", "SELF", "PARTNER", "BLACKLIST"}[m]
}
func (m PeerState) EnumIndex() int {
return int(m)
}
type PeerRelation int type PeerRelation int
const ( const (
@@ -69,7 +53,6 @@ type Peer struct {
NATSAddress string `json:"nats_address" bson:"nats_address" validate:"required"` NATSAddress string `json:"nats_address" bson:"nats_address" validate:"required"`
WalletAddress string `json:"wallet_address" bson:"wallet_address" validate:"required"` // WalletAddress is the wallet address of the peer WalletAddress string `json:"wallet_address" bson:"wallet_address" validate:"required"` // WalletAddress is the wallet address of the peer
PublicKey string `json:"public_key" bson:"public_key" validate:"required"` // PublicKey is the public key of the peer PublicKey string `json:"public_key" bson:"public_key" validate:"required"` // PublicKey is the public key of the peer
State PeerState `json:"state" bson:"state" default:"0"`
Relation PeerRelation `json:"relation" bson:"relation" default:"0"` Relation PeerRelation `json:"relation" bson:"relation" default:"0"`
ServicesState map[string]int `json:"services_state,omitempty" bson:"services_state,omitempty"` ServicesState map[string]int `json:"services_state,omitempty" bson:"services_state,omitempty"`
FailedExecution []PeerExecution `json:"failed_execution" bson:"failed_execution"` // FailedExecution is the list of failed executions, to be retried FailedExecution []PeerExecution `json:"failed_execution" bson:"failed_execution"` // FailedExecution is the list of failed executions, to be retried

View File

@@ -61,9 +61,6 @@ func (a *peerMongoAccessor) GetObjectFilters(search string) *dbs.Filters {
search = "" search = ""
} }
return &dbs.Filters{ return &dbs.Filters{
And: map[string][]dbs.Filter{ // search by name if no filters are provided
"state": {{Operator: dbs.EQUAL.String(), Value: ONLINE.EnumIndex()}},
},
Or: map[string][]dbs.Filter{ // search by name if no filters are provided Or: map[string][]dbs.Filter{ // search by name if no filters are provided
"abstractobject.name": {{Operator: dbs.LIKE.String(), Value: search}}, "abstractobject.name": {{Operator: dbs.LIKE.String(), Value: search}},
"url": {{Operator: dbs.LIKE.String(), Value: search}}, "url": {{Operator: dbs.LIKE.String(), Value: search}},

View File

@@ -64,8 +64,6 @@ func (dca *ResourceMongoAccessor[T]) UpdateOne(set utils.DBObject, id string) (u
if d, c, err := utils.GenericUpdateOne(set, id, dca, dca.generateData()); err != nil { if d, c, err := utils.GenericUpdateOne(set, id, dca, dca.generateData()); err != nil {
return d, c, err return d, c, err
} else { } else {
d.Unsign()
d.Sign()
return utils.GenericUpdateOne(set, id, dca, dca.generateData()) return utils.GenericUpdateOne(set, id, dca, dca.generateData())
} }
} }
@@ -75,11 +73,7 @@ func (dca *ResourceMongoAccessor[T]) StoreOne(data utils.DBObject) (utils.DBObje
return nil, 404, errors.New("can't create a non existing computing units resource not reported onto compute units catalog") return nil, 404, errors.New("can't create a non existing computing units resource not reported onto compute units catalog")
} }
data.(T).Trim() data.(T).Trim()
d, c, err := utils.GenericStoreOne(data, dca) return utils.GenericStoreOne(data, dca)
if err != nil {
return d, c, err
}
return dca.UpdateOne(d, d.GetID())
} }
func (dca *ResourceMongoAccessor[T]) CopyOne(data utils.DBObject) (utils.DBObject, int, error) { func (dca *ResourceMongoAccessor[T]) CopyOne(data utils.DBObject) (utils.DBObject, int, error) {

View File

@@ -99,8 +99,10 @@ func GenericUpdateOne(set DBObject, id string, a Accessor, new DBObject) (DBObje
} }
set = newSet set = newSet
r.UpToDate(a.GetUser(), a.GetPeerID(), false) r.UpToDate(a.GetUser(), a.GetPeerID(), false)
r.Unsign() if a.GetPeerID() == r.GetCreatorID() {
r.Sign() r.Unsign()
r.Sign()
}
if a.ShouldVerifyAuth() && !r.VerifyAuth("update", a.GetRequest()) { if a.ShouldVerifyAuth() && !r.VerifyAuth("update", a.GetRequest()) {
return nil, 403, errors.New("you are not allowed to access :" + a.GetType().String()) return nil, 403, errors.New("you are not allowed to access :" + a.GetType().String())
} }

View File

@@ -14,19 +14,20 @@ import (
) )
type NATSResponse struct { type NATSResponse struct {
FromApp string `json:"from_app"` FromApp string `json:"from_app"`
Datatype DataType `json:"datatype"` Datatype DataType `json:"datatype"`
User string `json:"user"` User string `json:"user"`
Method int `json:"method"` Method int `json:"method"`
Payload []byte `json:"payload"` SearchAttr string `json:"search_attr"`
Payload []byte `json:"payload"`
} }
// NATS Method Enum defines the different methods that can be used to interact with the NATS server // NATS Method Enum defines the different methods that can be used to interact with the NATS server
type NATSMethod int type NATSMethod int
var meths = []string{"remove execution", "create execution", "discovery", var meths = []string{"remove execution", "create execution", "discovery",
"workflow event", "remove peer", "create peer", "create resource", "remove resource", "verify_resource", "workflow event", "create resource", "remove resource",
"propalgation event", "catalogsearch event", "propalgation event", "catalog search event",
} }
const ( const (
@@ -35,12 +36,8 @@ const (
DISCOVERY DISCOVERY
WORKFLOW_EVENT WORKFLOW_EVENT
REMOVE_PEER
CREATE_PEER
CREATE_RESOURCE CREATE_RESOURCE
REMOVE_RESOURCE REMOVE_RESOURCE
VERIFY_RESOURCE
PROPALGATION_EVENT PROPALGATION_EVENT
CATALOG_SEARCH_EVENT CATALOG_SEARCH_EVENT
@@ -53,7 +50,7 @@ func (n NATSMethod) String() string {
// NameToMethod returns the NATSMethod enum value from a string // NameToMethod returns the NATSMethod enum value from a string
func NameToMethod(name string) NATSMethod { func NameToMethod(name string) NATSMethod {
for _, v := range [...]NATSMethod{REMOVE_EXECUTION, CREATE_EXECTUTION, DISCOVERY, WORKFLOW_EVENT, for _, v := range [...]NATSMethod{REMOVE_EXECUTION, CREATE_EXECTUTION, DISCOVERY, WORKFLOW_EVENT,
REMOVE_PEER, CREATE_PEER, CREATE_RESOURCE, REMOVE_RESOURCE, VERIFY_RESOURCE, PROPALGATION_EVENT, CATALOG_SEARCH_EVENT} { CREATE_RESOURCE, REMOVE_RESOURCE, PROPALGATION_EVENT, CATALOG_SEARCH_EVENT} {
if strings.Contains(strings.ToLower(v.String()), strings.ToLower(name)) { if strings.Contains(strings.ToLower(v.String()), strings.ToLower(name)) {
return v return v
} }