cb073786b3
Enthält: - rdp_client.py: RDP Client mit GUI und Monitor-Auswahl - rdp.sh: Bash-basierter RDP Client - teamleader_test/: Network Scanner Fullstack-App - teamleader_test2/: Network Mapper CLI Subdirectories mit eigenem Repo wurden ausgeschlossen. Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
123 lines
4.9 KiB
Python
123 lines
4.9 KiB
Python
"""SQLAlchemy database models."""
|
|
|
|
from sqlalchemy import Column, Integer, String, DateTime, Float, Text, ForeignKey, Table, JSON
|
|
from sqlalchemy.orm import relationship
|
|
from datetime import datetime
|
|
|
|
from app.database import Base
|
|
|
|
|
|
# Association table for many-to-many relationship between scans and hosts
|
|
scan_hosts = Table(
|
|
'scan_hosts',
|
|
Base.metadata,
|
|
Column('scan_id', Integer, ForeignKey('scans.id', ondelete='CASCADE'), primary_key=True),
|
|
Column('host_id', Integer, ForeignKey('hosts.id', ondelete='CASCADE'), primary_key=True)
|
|
)
|
|
|
|
|
|
class Scan(Base):
|
|
"""Model for scan operations."""
|
|
|
|
__tablename__ = 'scans'
|
|
|
|
id = Column(Integer, primary_key=True, index=True)
|
|
started_at = Column(DateTime, nullable=False, default=datetime.utcnow)
|
|
completed_at = Column(DateTime, nullable=True)
|
|
scan_type = Column(String(50), nullable=False, default='quick')
|
|
network_range = Column(String(100), nullable=False)
|
|
status = Column(String(20), nullable=False, default='pending')
|
|
hosts_found = Column(Integer, default=0)
|
|
ports_scanned = Column(Integer, default=0)
|
|
error_message = Column(Text, nullable=True)
|
|
|
|
# Relationships
|
|
hosts = relationship('Host', secondary=scan_hosts, back_populates='scans')
|
|
|
|
def __repr__(self) -> str:
|
|
return f"<Scan(id={self.id}, network={self.network_range}, status={self.status})>"
|
|
|
|
|
|
class Host(Base):
|
|
"""Model for discovered network hosts."""
|
|
|
|
__tablename__ = 'hosts'
|
|
|
|
id = Column(Integer, primary_key=True, index=True)
|
|
ip_address = Column(String(45), nullable=False, unique=True, index=True)
|
|
hostname = Column(String(255), nullable=True)
|
|
mac_address = Column(String(17), nullable=True)
|
|
first_seen = Column(DateTime, nullable=False, default=datetime.utcnow)
|
|
last_seen = Column(DateTime, nullable=False, default=datetime.utcnow, onupdate=datetime.utcnow)
|
|
status = Column(String(20), nullable=False, default='online', index=True)
|
|
os_guess = Column(String(255), nullable=True)
|
|
device_type = Column(String(50), nullable=True)
|
|
vendor = Column(String(255), nullable=True)
|
|
notes = Column(Text, nullable=True)
|
|
|
|
# Relationships
|
|
services = relationship('Service', back_populates='host', cascade='all, delete-orphan')
|
|
scans = relationship('Scan', secondary=scan_hosts, back_populates='hosts')
|
|
outgoing_connections = relationship(
|
|
'Connection',
|
|
foreign_keys='Connection.source_host_id',
|
|
back_populates='source_host',
|
|
cascade='all, delete-orphan'
|
|
)
|
|
incoming_connections = relationship(
|
|
'Connection',
|
|
foreign_keys='Connection.target_host_id',
|
|
back_populates='target_host',
|
|
cascade='all, delete-orphan'
|
|
)
|
|
|
|
def __repr__(self) -> str:
|
|
return f"<Host(id={self.id}, ip={self.ip_address}, hostname={self.hostname})>"
|
|
|
|
|
|
class Service(Base):
|
|
"""Model for services running on hosts (open ports)."""
|
|
|
|
__tablename__ = 'services'
|
|
|
|
id = Column(Integer, primary_key=True, index=True)
|
|
host_id = Column(Integer, ForeignKey('hosts.id', ondelete='CASCADE'), nullable=False)
|
|
port = Column(Integer, nullable=False)
|
|
protocol = Column(String(10), nullable=False, default='tcp')
|
|
state = Column(String(20), nullable=False, default='open')
|
|
service_name = Column(String(100), nullable=True)
|
|
service_version = Column(String(255), nullable=True)
|
|
banner = Column(Text, nullable=True)
|
|
first_seen = Column(DateTime, nullable=False, default=datetime.utcnow)
|
|
last_seen = Column(DateTime, nullable=False, default=datetime.utcnow, onupdate=datetime.utcnow)
|
|
|
|
# Relationships
|
|
host = relationship('Host', back_populates='services')
|
|
|
|
def __repr__(self) -> str:
|
|
return f"<Service(host_id={self.host_id}, port={self.port}, service={self.service_name})>"
|
|
|
|
|
|
class Connection(Base):
|
|
"""Model for detected connections between hosts."""
|
|
|
|
__tablename__ = 'connections'
|
|
|
|
id = Column(Integer, primary_key=True, index=True)
|
|
source_host_id = Column(Integer, ForeignKey('hosts.id', ondelete='CASCADE'), nullable=False, index=True)
|
|
target_host_id = Column(Integer, ForeignKey('hosts.id', ondelete='CASCADE'), nullable=False, index=True)
|
|
connection_type = Column(String(50), nullable=False)
|
|
protocol = Column(String(10), nullable=True)
|
|
port = Column(Integer, nullable=True)
|
|
confidence = Column(Float, nullable=False, default=1.0)
|
|
detected_at = Column(DateTime, nullable=False, default=datetime.utcnow)
|
|
last_verified = Column(DateTime, nullable=True)
|
|
extra_data = Column(JSON, nullable=True)
|
|
|
|
# Relationships
|
|
source_host = relationship('Host', foreign_keys=[source_host_id], back_populates='outgoing_connections')
|
|
target_host = relationship('Host', foreign_keys=[target_host_id], back_populates='incoming_connections')
|
|
|
|
def __repr__(self) -> str:
|
|
return f"<Connection(source={self.source_host_id}, target={self.target_host_id}, type={self.connection_type})>"
|