105 lines
4.3 KiB
Python
105 lines
4.3 KiB
Python
import json
|
|
from datetime import datetime, timezone
|
|
from sqlalchemy import Boolean, Column, DateTime, Float, Integer, String, Text
|
|
from app.database import Base
|
|
|
|
|
|
def _now():
|
|
return datetime.now(timezone.utc)
|
|
|
|
|
|
class NetObjectModel(Base):
|
|
__tablename__ = "net_objects"
|
|
id = Column(String, primary_key=True, index=True)
|
|
name = Column(String, nullable=False)
|
|
slug = Column(String, nullable=True)
|
|
type = Column(String, nullable=False)
|
|
parent_id = Column(String, nullable=True, index=True)
|
|
status = Column(String, nullable=False, default="active")
|
|
vendor = Column(String, nullable=True)
|
|
model = Column(String, nullable=True)
|
|
serial = Column(String, nullable=True)
|
|
notes = Column(Text, nullable=True)
|
|
created_at = Column(DateTime(timezone=True), default=_now)
|
|
updated_at = Column(DateTime(timezone=True), default=_now, onupdate=_now)
|
|
|
|
|
|
class RelationModel(Base):
|
|
__tablename__ = "relations"
|
|
id = Column(String, primary_key=True, index=True)
|
|
type = Column(String, nullable=False)
|
|
layer = Column(String, nullable=False)
|
|
source_id = Column(String, nullable=False, index=True)
|
|
target_id = Column(String, nullable=False, index=True)
|
|
direction = Column(String, nullable=False, default="forward")
|
|
source = Column(String, nullable=False)
|
|
confidence = Column(Float, nullable=False, default=1.0)
|
|
status = Column(String, nullable=False, default="active")
|
|
attributes_json = Column(Text, nullable=True, default="{}")
|
|
created_at = Column(DateTime(timezone=True), default=_now)
|
|
updated_at = Column(DateTime(timezone=True), default=_now, onupdate=_now)
|
|
|
|
@property
|
|
def attributes(self):
|
|
return json.loads(self.attributes_json or "{}")
|
|
|
|
@attributes.setter
|
|
def attributes(self, value):
|
|
self.attributes_json = json.dumps(value)
|
|
|
|
|
|
class JobModel(Base):
|
|
__tablename__ = "jobs"
|
|
id = Column(String, primary_key=True, index=True)
|
|
type = Column(String, nullable=False)
|
|
status = Column(String, nullable=False, default="pending")
|
|
progress = Column(Integer, nullable=False, default=0)
|
|
message = Column(Text, nullable=True)
|
|
error = Column(Text, nullable=True)
|
|
payload_json = Column(Text, nullable=True)
|
|
result_json = Column(Text, nullable=True)
|
|
created_at = Column(DateTime(timezone=True), default=_now)
|
|
updated_at = Column(DateTime(timezone=True), default=_now, onupdate=_now)
|
|
started_at = Column(DateTime(timezone=True), nullable=True)
|
|
finished_at = Column(DateTime(timezone=True), nullable=True)
|
|
|
|
|
|
class DiscoveryProfileModel(Base):
|
|
__tablename__ = "discovery_profiles"
|
|
id = Column(String, primary_key=True, index=True)
|
|
name = Column(String, nullable=False)
|
|
strategy = Column(String, nullable=False, default="fast")
|
|
use_icmp = Column(Boolean, nullable=False, default=True)
|
|
use_arp = Column(Boolean, nullable=False, default=True)
|
|
use_snmp = Column(Boolean, nullable=False, default=False)
|
|
use_lldp = Column(Boolean, nullable=False, default=False)
|
|
use_portscan = Column(Boolean, nullable=False, default=False)
|
|
notes = Column(Text, nullable=True)
|
|
|
|
|
|
class DiscoveryRunModel(Base):
|
|
__tablename__ = "discovery_runs"
|
|
id = Column(String, primary_key=True, index=True)
|
|
profile_id = Column(String, nullable=False, index=True)
|
|
subnet_ids_json = Column(Text, nullable=False, default="[]")
|
|
job_id = Column(String, nullable=True)
|
|
status = Column(String, nullable=False, default="pending")
|
|
summary_json = Column(Text, nullable=True)
|
|
created_at = Column(DateTime(timezone=True), default=_now)
|
|
updated_at = Column(DateTime(timezone=True), default=_now, onupdate=_now)
|
|
started_at = Column(DateTime(timezone=True), nullable=True)
|
|
finished_at = Column(DateTime(timezone=True), nullable=True)
|
|
|
|
|
|
class UserModel(Base):
|
|
__tablename__ = "users"
|
|
id = Column(Integer, primary_key=True, autoincrement=True)
|
|
username = Column(String, unique=True, nullable=False, index=True)
|
|
email = Column(String, unique=True, nullable=True)
|
|
hashed_password = Column(String, nullable=False)
|
|
role = Column(String, nullable=False, default="viewer")
|
|
disabled = Column(Boolean, nullable=False, default=False)
|
|
force_password_change = Column(Boolean, nullable=False, default=False)
|
|
created_at = Column(DateTime(timezone=True), default=_now)
|
|
updated_at = Column(DateTime(timezone=True), default=_now, onupdate=_now)
|