app.py

import os 
from flask import Flask, render_template, request, redirect, url_for, flash, abort, Response
import sqlite3
import ssl
import subprocess

app = Flask(__name__)

import sqlite3
from flask import Flask, render_template, request, redirect, url_for, session

app = Flask(__name__)
app.secret_key = "1234"

# DB 초기화
def init_db():
    connection = None
    try:
        # DB 파일이 없으면 생성
        connection = sqlite3.connect("webDB") 
        cursor = connection.cursor()
        
        # User 테이블 생성
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS users (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                username TEXT NOT NULL UNIQUE,
                password TEXT NOT NULL
            );
        ''')

        # Post 테이블 생성 
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS posts (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                title TEXT NOT NULL,
                content TEXT NOT NULL,
                author TEXT NOT NULL
            );
        ''')
        
        # admin 계정 삽입 
        cursor.execute("INSERT OR IGNORE INTO users (username, password) VALUES ('admin', 'admin');")

        # 게시물 데이터 삽입
        cursor.execute("INSERT OR IGNORE INTO posts (id, title, content, author) VALUES (1, '테스트 게시글1', '첫 번째 글입니다.', '게시자 1');")
        cursor.execute("INSERT OR IGNORE INTO posts (id, title, content, author) VALUES (2, '테스트 게시글2', '두 번째 글입니다.', '게시자 2');")
        cursor.execute("INSERT OR IGNORE INTO posts (id, title, content, author) VALUES (3, '테스트 게시글3', '세 번째 글입니다.', '게시자 3');")
        cursor.execute("INSERT OR IGNORE INTO posts (id, title, content, author) VALUES (4, '테스트 게시글4', '네 번째 글입니다.', '게시자 4');")

        
        connection.commit()
        print("DB 초기화 성공")
        
    except Exception as e:
        print(f"DB 초기화 실패: {e}")
        
    finally:
        if connection:
            connection.close()

# DB 연결 함수 
def get_db_connection():
    conn = sqlite3.connect("webDB")
    conn.row_factory = sqlite3.Row 
    return conn

def test():
    return 

@app.route('/')
def index():
    conn = get_db_connection()
    posts = conn.execute("SELECT * FROM posts ORDER BY id DESC").fetchall()
    conn.close()
    return render_template('index.html', posts=posts)

# SQLi와 brute forcing 취약점 
@app.route('/login', methods=['GET', 'POST'])
def login():
    if request.method == 'POST':
        username = request.form['username']
        password = request.form['password']
        
        conn = get_db_connection()
        cursor = conn.cursor()

        query = "SELECT * FROM users WHERE username = '" + username + "' AND password = '" + password + "'"
        
        try:
            cursor.execute(query)
            user = cursor.fetchone() 
            
            if user:
                session['username'] = user['username']
                return redirect(url_for('index'))
            else:
                return render_template('login.html', error="유효하지 않은 사용자 이름 또는 비밀번호입니다.")

        except Exception as e:
            print(f"로그인 중 오류 발생: {e}")
            return render_template('index.html', error="로그인 처리 중 오류가 발생했습니다.")
        
        finally:
            conn.close() 
    return render_template('login.html')

# 로그아웃
@app.route('/logout')
def logout():
    session.pop('username', None)
    return redirect(url_for('index'))

#  StoredXSS 취약점
@app.route('/post', methods=['GET', 'POST'])
def createPost():
    if 'username' not in session:
        return redirect(url_for('login'))

    if request.method == 'POST':
        title = request.form['title']
        content = request.form['content']
        author = session['username']

        conn = get_db_connection()
        cursor = conn.cursor()
        query = f"INSERT INTO posts (title, content, author) VALUES ('{title}', '{content}', '{author}')"
        cursor.execute(query)

        conn.commit()
        conn.close()

        return redirect(url_for('index'))  # 글 작성 후 메인으로

    return render_template('post.html')

# command-injection 취약점
@app.route('/ping', methods=['GET', 'POST'])
def ping():
    if 'username' not in session:
        return redirect(url_for('login'))

    if request.method == 'POST':
        host = request.form.get('host')
        cmd = f'ping -c 3 "{host}"'
        try:
            output = subprocess.check_output(['/bin/sh', '-c', cmd], timeout=5)
            return render_template('ping_result.html', data=output.decode('utf-8'))
        except subprocess.TimeoutExpired:
            return render_template('ping_result.html', data='Timeout !')
        except subprocess.CalledProcessError:
            return render_template('ping_result.html', data=f'an error occurred while executing the command. -> {cmd}')

    return render_template('ping.html')


if __name__ == '__main__':
    init_db() 
    # app.run(host="0.0.0.0", port=5000)
    app.run(host='0.0.0.0', ssl_context='adhoc', port=8443)
    # app.run(host='0.0.0.0', port=8443)